Another question, is it possible to have an already-running Flow Run update when the flow itself gets updated? Suppose a task is added, or modified, or removed? Or to cancel a Flow Run and then have it restart from cancelled, but the new run would include the recent updates to the flow?

Im setting up an automation for "When any run from <MyFlowName> changes to Failed, then <<My Action>>". But when I save the automation... it changes to "When any run from <MyFlowName> changes to any of these states, then <<My Action>>". is this an issue with the GUI.. or is my automation not saving the correct values in the wizard?

I'm experiencing a weird issue when I register my flows • I had only one flow with this dir structure ◦

extract_data.py

flow file (config for the flow) ▪︎ here I import the tasks directory with a simple

import tasks

◦

tasks

at the same level, a directory with my tasks ◦ When I register this flow it works correctly in schedule basis • I just added a new flow

calculate_something.py

• The tasks for this flow are in the

tasks

directory too • now when I want to run this new flow I get

FlowStorageError('An error occurred while unpickling the flow:\n ModuleNotFoundError("No module named \'tasks\'")

What I don't get is... Why is it working for the first flow and not for this new one? Any ideas? I appreciate it

Hi, I have a flow like this A -> List -> Dict -> B, Result of A is persisted in S3, Whenever B fails and i try to restart the flow Dict task supplies empty data to B since its result is not persisted. Dict is an internal task which we don’t have any control, How we can overcome this?

Hi, we are currently evaluation Prefect Orion (and we are pretty amazed to be honest). One challenge I cannot wrap my had around at the moment is manual interaction (e.g., "wait for approval"). We are crunching large visual data sets that often have unexpected features. Because of this, we need some kind of manual sanity check by a human expert before we start the compute/time intensive tasks. In Prefect 1.0 this was possible via the

manual_only

trigger. Is it possible to do the same with Orion (in the future)?

Hello, I would like to upgrade my prefect server in my k8s cluster to prefect orion. Is there any guidelines on how to do this?

Hello, I am a bit confused about how task results are stored when I use kubernetes. Can I define a mount to network storage (in the job template) and then use local storage so it is saved there? Would server be able to read it (assuming is also mounted on the same network storage)? Alternatively, we have sonatype nexus repository - can I create a result type based on that (implemented the interface?) If I do so, how can I ensure this code is available to the system (task / server)

FYI: if you face some issues with Slack today, you can post your question on Discourse or GitHub. S status is all green but e.g. search doesn't seem to work today.

Hi. I am trying to add checkpointing in my flows to checkpoint Dask dataframe that I pass between tasks: Tasks:

@task(result=S3Result(bucket='bucket-name'))
def checkpoint_data() -> dd.DataFrame:
    df = pd.DataFrame({'col_1': ['1', '2', '3'], 'col_2': [1, 2, 3]})
    ddf = dd.from_pandas(df, npartitions=1)

    return ddf


@task()
def accept_checkpointed_data(ddf: dd.DataFrame) -> None:
    raise ValueError("Checkpoint testing...")

Flow:

ddf = checkpoint_data()
accept_checkpointed_data(ddf)

How to be sure that on restart after failure(ValueError I added) accept_checkpointed_data task loads data from S3 and not using cached_inputs? Thanks

Hi! Using

PrefectSecret()

to load secrets from Prefect Cloud. Is it possible to load secret from cloud into

prefect.context.secrets

during flow start so all tasks can access them when needed? Most of our tasks variables are auth-related, and the code/layout is getting overly complex.

Hello, for a prefect task (which queries mysql database with pymysql) I am getting this error, is there a way to increase the timeout ?

No heartbeat detected from the remote task; marking the run as failed.

Hey do you have any idea of a prefect prometheus exporter which goal was for example to monitor unhealthy agents?

Hello, is it possible to dynamically set which

project

a

create_flow_run

call creates a Flow for depending on environment variables? Example in thread

Hi all! Just a heads up, no PrefectLive this week. Join the #events channel to keep tabs on livestreams! https://prefect-community.slack.com/archives/C036FRC4KMW/p1646843588906869

Is there any way to limit flow concurrency without affecting which agents a flow can be deployed to?

Second question, I have a mapped flow pattern like below. How can I ensure this kicks off all sub-flows simultaneously before waiting for any individual run to complete? I am using a synchronous LocalDaskExecutor for this.

child_ids_result = create_flow_run.map(
    flow_name=unmapped(...),
    project_name=unmapped(...),
    parameters=parameters,
    run_name=run_names,
)
wait_for_flow_run_result = wait_for_flow_run.map(
    flow_run_id=child_ids_result,
    stream_states=unmapped(True),
    stream_logs=unmapped(True),
    raise_final_state=unmapped(True),
)
final_task(..., upstream_tasks=[wait_for_flow_run_result])

new here and trying this vs airflow for a side project! Can I have classes or methods defined in other files that can be referenced as tasks in a prefect flow? For example, lets say I have this file structure:

scrapers
> scraper1
> scraper2
prefect_flow.py

and scraper1/2 have Classes or methods that scrape specific websites. Am I able to leave those defined in those files, while referencing them in my flow.py file? It would look something like this..

import ...

@task 
def scrape_site_one():
    s = Scraper()
    return s.scrape_site('www...')
....

with Flow("update historical pricing") as flow:
    scrape_site_one()

Hi! Since yesterday one of our flows has stopped running, it is marked as white in the UI (see screenshot). This is pretty urgent so it would be nice if I could get some quick help!

Hi, Trying to understand how I can use prefect with dask while still debug locally I am getting an error. More details, code and errors inside thread

Hello! I’m new here and evaluating a switch from Airflow to Prefect. We’re not doing any processing in Python/DataFrames/etc. but are using GCP with Beam/Dataflow (don’t see any specific Prefect integration, but I assume it can be easily be done by packaging it in a container and running with a RunNamespacedJob task or similar), BigQuery, Avro on GCS and DBT. Almost all examples I see using cloud storage works with pickled dataframes that are pulled locally when executing the task. I’m looking to use a setup that references Avro datasets on GCS and/or BQ tables and passes them as in/outputs to tasks running as external services (Dataflow jobs, BQ queries, DBT models), similar to Luigi’s Target class. Given Prefect’s focus on dataflow I would expect that to be a supported pattern but haven’t yet been able to find any guides/tutorials/discussions about this, does anyone have any tips? Preferably for Orion as we’re just getting started, but otherwise for 1.0 as a reference. Thanks a lot!

Does anyone have advice for docker agents to stay logged in on the Azure ACR? I have the docker agents running on my vm, it's logged in with an identity, but the

az acr login --name acrname

logs out after a couple hours and requires me to do it again. I also run the command within the docker image as well (for the docker agent) It's all working, but after a while I get logged out from the ACR and my flow starts to crash.

Error message:

500 Server Error for http+docker://localhost/v1.41/images/create?tag=latest&amp;fromImage=acrname.azurecr.io%2Fimage: Internal Server Error ("Head https://acrname.azurecr.io/v2/image/manifests/latest: unauthorized: authentication required")

Hi, A general question about how dask is used under the hood. In "regular" dask code there's usually a client (script or from REPL) which connects to a cluster (ephemeral or long-running) and then it can submit work to that cluster. How this is done in Prefect? Assuming I am using DaskExecutor and running on k8s with kubernetes agent - does the client code run in the agents itself? As part of the KubeCluster? Where should I look for the code running it - k8s agent?

Hey guys what is standard way to restart flows via client (library)? I'm trying set to

set_flow_run_state

Scheduled (i tried to Pending but this make flow be in idle and not be executed) i want to restart all the tasks.

client.set_flow_run_state(
        flow_run_id=flow_run_id,
        state=Scheduled()
)

Hey everyone! Quick question (sorry if this has been posted but could not find any answers out there). Is it possible to use the new Prefect Orion together with Airbyte for streaming data pipelines? Or does it only work with the standard Prefect product?

Has anyone encountered the following error when using the ECS agent? Is there an easy way to increase the number of retries or backoff interval?

An error occurred (ThrottlingException) when calling the RegisterTaskDefinition operation (reached max retries: 2): Rate exceeded

How can I run two tasks (one that is mapped, the other that is not) in parallel?

@task
def sleep_5():
    print(f'sleeping 5')
    time.sleep(5)
    print(f'done sleeping 5')


@task
def sleep_x(x):
    print(f'sleeping {x}')
    time.sleep(x)
    print(f'done sleeping {x}')


with DHFlow("foo") as flow:
    sleep_5()
    sleep_x.map([7, 8])

when I execute the above,

sleep_5

is run first; only after it finishes will

sleep_x

run.

Hi there. I have a flow that is orchestrated by Prefect Cloud. It has 1k+ mapped tasks and it failed after processing 250. There are 20 tasks in retrying state showing this message:

No heartbeat detected from the remote task; retrying the run.This will be retry 1 of 2.

However, the last message was written 12 hours ago. It does not look like the flow got retried at all. Could someone help me figure out what may have happened? I am using ECS to run the flows. The flow run is

fed29ba9-66a6-4f0b-aa69-cf9db908bd58

I went ahead and canceled it but it does not offer me the option to restart it. How can I restart it and preserve the same flow run ID so that the tasks that succeeded are not executed again? Ideally, I'd rely on the task state but if they need to run again, I am hoping that caching will help avoid doing the work again. The cache key is based on the

flow_run_id

. Thanks!

Hi, I'm running

n_workers

agents as kubernetes jobs similarly to this:

# Agent flow
with Flow("run-agent") as flow_agent:
    ShellTask(agent_command) # launch agent computing tasks

@task
def run_agent(worker_id): 
    run_agent_flow_id = create_flow_run.run("run-agent",
                                            name=f"run-agent_#{worker_id}")
    return run_agent_flow_id

# Main flow
with Flow("main") as flow:
    n_workers = Parameters("n_workers", default=2)
    worker_ids = get_worker_ids(n_workers)  # [0, 1]
    run_agent.map(worker_id=worker_ids)

When the agents finish their tasks, for some reason the kubernetes jobs are not terminating. Inside the job's container apparently the agent's process terminates, but I see

prefect execute flow-run

and

prefect hearbeat flow-run -i ...

being stuck.

Hi Community! Are there guidelines to free up the previous tasks' memory after they are finished? We are running memory intensive pipelines and run into out of memory error. The output of each task is only needed in the next task, but not in the subsequent tasks. Is there room for memory optimization?

I am pushing my code to github then want to build it in a docker image and lastly want to run the prefect pipeline as part of our kubernetes orchestration that we already have set up. When I push my code, I would need to build it and then push it to azure container registry. Then, in kubernetes, I need to run a prefect flow like this one https://github.com/anna-geller/packaging-prefect-flows/blob/master/flows/docker_pickle_kubernetes_run_custom_ecr_image.py?