Hi all, Bear with me on this, but I'm a pretty new data engineer with a few Prefect flows under my belt, and have a question on best practices with regards to model deployment: Current setup is this: Agent is on AKS, file storage on Blob (we're an Azure shop). Using Prefect Cloud. What I'm trying to do: I have a model that has a child folder called Data. I'm trying to run this model using Prefect. Basically, I need to call a few API's and save the .csv's in the Data folder, run the model that takes input from the newly saved files, output some new files into the Data folder, and then I need to do some other transformations on a couple of the outputted files prior to snowflake upload. I've so far thrown the entire model folder system into the Flows folder, deployed the flow to Blob Storage, and attempted to run the model using a

poetry run python src/driver/__main__.py

shell_run_command. This has been a slow process as there's been a bunch of little issues arising, etc. My question: What would be the easiest way of doing all this? ie. what would be the best method to approach this problem? I feel like my method is a bit... patchy?

Hi all, is there a way to use prefect parameter in @contextlib.contextmanager? we have a function that's get's Snowflake credentials from AWS secrets, but at run time, we want to allow testers to be able to enter the Snowflake schema they like. so I tried to put a parameter for the schema in the flow run but it's not working.

Has anybody migrated from 1.x to Prefect 2? And how was the journey?

Is there an easy way to disable caching when running tests? Something like the

disable_run_logger

Hi! Is there a way to concurrency limit a task running with

.map

? I've tried using CLI

prefect concurrency-limit

tagging the task but apperently there was no effect and the limit was not respected

i'm trying to deploy the Prefect agent to GKE. I need a specific version of Python (3.9.13) on the image so I changed the Dockerfile as follows:

FROM python:3.9.13

COPY requirements.txt .

RUN pip install -r requirements.txt

Prefect is included in the requirements. When the job is launched, I get the following error:

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/prefect/engine.py", line 247, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/usr/local/lib/python3.9/site-packages/prefect/client/utilities.py", line 47, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/prefect/deployments.py", line 159, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=deployment.path, local_path=".")
  File "/usr/local/lib/python3.9/site-packages/prefect/filesystems.py", line 144, in get_directory
    shutil.copytree(from_path, local_path, dirs_exist_ok=True)
  File "/usr/local/lib/python3.9/shutil.py", line 566, in copytree
    with os.scandir(src) as itr:
FileNotFoundError: [Errno 2] No such file or directory: '/opt/prefect/flows'

How do I resolve this?

Let's say I want to use

run_deployment()

in a flow, and I want to run multiple deployments concurrently, then move on to the next task once all the deployments have finished running. How might I do that?

What is the expected behaviour when two Prefect agents are listening to the same work queue? We have run an experiment and found that one Agent runs the flows while the other does no work - is this expected? Will the agents ever share work between them?

Hi, I'm still evaluating if Prefect is the right tool for our team and I have a question about pricing. In the

Account

tab it says first 20k task runs are free and the pricing page that is linked there doesn't mention anything at all about number of task runs. Could someone help me understand please how pricing works? The descriptions of our plan in the

Account

tab and the plans in the pricing page are completely different. I would love to make Prefect work for us but I don't have an approval for the paid version until I'm confident that Prefect is the right tool for us. With the current setup we have plenty of issues and I'd love to break down the current tasks into smaller ones. However, with that I'd most likely go over the 20k monthly limit already at this point even before we start growing.

I use Prefect Cloud. A few days ago I lost my workspace and all my profile data disappeared. But the previously configured flows are working on schedule, although I don't see their statuses in the UI. Has anyone encountered something like this?

Hi, I migrated from prefect v1 to prefect orion (i love the prefect orion cli 😄 thanks for that !). I have a lot of questions to ask, but I'll start with this : • Agents are not supervised? (with heartbeat like in prefectv1) • When an agent is lost (network issue, oomkill .. and more) the middle of a run, the flow is still in "running" state, is this normal behavior? how to detect this problem? Thanks for all, Matt

Hi - I am getting permission issues when logged into the UI - clicking either of my two workspaces gives a 404 - the {workspace}/permissions call is returning

{"detail":"Not Found"}

how can I setup SSO in prefect cloud 2.0?

is there a way to force stop a flow run ?

Currently, the ability to use custom kube config for

KubernetesJob

does not seem to be documented in the source, docs, or api docs, which would lead one to believe that it's not possible, even though it is. Just FYI.

Hi, how could I run a job every X times until a condition ?

my use case is buy domain name (OVH api) + waiting for custom NS propagation

Hi - When deploying a flow with a S3 block and Docker infrastructure block, will the flow be looking for the code to run in the docker container, or S3 block? The deployment is sending our code to S3, but I'm having issues running the flow. It appears to be looking for the code in the docker image. CLI Deployment script:

prefect deployment build -n fidelity-allocations-deployment -q default -sb s3/prefect-training -ib docker-container/prefect-training -a src/main/prefect/flows/allocations/prefect_fidelity_allocations.py:FidelityAllocationsFlow --cron "13 21 * * 1-5"

Error message from the flow: Flow could not be retrieved from deployment. Traceback (most recent call last): File "<frozen importlib._bootstrap_external>", line 846, in exec_module File "<frozen importlib._bootstrap_external>", line 982, in get_code File "<frozen importlib._bootstrap_external>", line 1039, in get_data FileNotFoundError: [Errno 2] No such file or directory: 'src\\main\\prefect\\flows\\allocations\\prefect_fidelity_allocations.py' The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/engine.py", line 257, in retrieve_flow_then_begin_flow_run flow = await load_flow_from_flow_run(flow_run, client=client) File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/client/orion.py", line 82, in with_injected_client return await fn(*args, **kwargs) File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/deployments.py", line 70, in load_flow_from_flow_run flow = await run_sync_in_worker_thread(import_object, str(import_path)) File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/utilities/asyncutils.py", line 57, in run_sync_in_worker_thread return await anyio.to_thread.run_sync(call, cancellable=True) File "/opt/pysetup/.venv/lib/python3.9/site-packages/anyio/to_thread.py", line 31, in run_sync return await get_asynclib().run_sync_in_worker_thread( File "/opt/pysetup/.venv/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread return await future File "/opt/pysetup/.venv/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 867, in run result = context.run(func, *args) File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/utilities/importtools.py", line 193, in import_object module = load_script_as_module(script_path) File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/utilities/importtools.py", line 156, in load_script_as_module raise ScriptError(user_exc=exc, path=path) from exc prefect.exceptions.ScriptError: Script at 'src\\main\\prefect\\flows\\allocations\\prefect_fidelity_allocations.py' encountered an exception

Hey all, it seems the allow_failure() wrapper doesn’t work with the

DaskTaskRunner()

😢 It can’t pickle the task futures, which is sadly blocking our migration to 2.0. Have tried this both in Dask locally and also on k8s. Any ideas? MRE in 🧵

Does anybody have a best practice for storing large data as a

Result

in 2.6+ ? I am thinking of passing data between tasks, persisted into remote storage as parquet, for example (e.g. using spark)

Has anyone got an example of python deployments with KubernetesJob using infra_overrides - when running the job it keeps telling me "instance of JsonPatch expected", not sure what to pass in, prefect 2.6.2

Hey guys! I'm programmatically creating all my deployments and am wondering what's a nice way to create them idempotently. Meaning, I run the

create_deployments

script multiple times and overwrite the deployments (similar to blocks). I'm currently using the

.build_from_flow

and

.apply

methods but I keep running into warnings

Hi Team, is there a good documentation regarding Installation of Prefect V2 on AWS ECS ?

Hi, probably straightforward question on Prefect 1.x, how do you force a flow to fail if a task fails?

Hi team! I'm building out a deployment flow for our CI/CD purposes for k8s and was looking to use the the

KubernetesJob.job_from_file

function to read a yaml file with our full k8s manifest. The function works, however I noticed that we have fields that get overwritten with default values in this call for

_shortcut_customizations

in

build_job

, specifically the

namespace

and

image

properties. I know that I can use an

infrastructure_override

when setting up the deployment, however there should be option to simply read a yaml file with those fields.

Hi, I'm having some issues with the async python Orion client and what looks to be a cloudpickle error. Code/screenshots in thread.

(prefect 2.0, deployment on cloud) I am running into a situation where all tasks are completed successfully but the flow is still running, what could be causing this?

Hey All - In the process of upgrading to prefect 2, so far really loving it. Having a few issues though. I am trying to build a custom image to run our flows. Starting with

prefecthq/prefect:2.6.1-python3.10-conda

as a base image. Couple of questions: • When installing a python environment in the container does it mater what it is called? I see the image has a

base

env and a

prefect

env. Should packages be installed in

prefect

env of can I create a new one? • If I create a new env should the Dockerfile include a line to activate the correct env.? • Is there a recipe for creating a custom image from a conda base image? I looked around and did find one, but maybe I missed it?

One more question: we are trying to run flows in a docker container locally for testing. I see in the docs that: • You must configure remote Storage. Local storage is not supported for Docker. Does this mean that we need to setup remote storage to test running the flows in the container? I currently have something like the following:

deploy_import_1 = Deployment.build_from_flow(
    flow=import_1,
    name="import1",
    work_queue_name="imports",
    schedule=IntervalSchedule(
        interval=timedelta(hours=1),
        anchor_date=datetime(2021, 1, 1, 2, 30, tzinfo=pytz.UTC)
    ),
    infrastructure=DockerContainer(
        image="path_to_image/import-prefect:python3.10",
        image_pull_policy="NEVER",
        auto_remove=False
    )
)

Do I need to include remote storage to test? I am a bit unclear on all the pieces of the puzzle still.

Wanted to highlight the #show-us-what-you-got channel with our revamped Prefect Recipes by @Serina and a KV/Store built by @Zanie 🎉