actually from the docs, I don't understand executor, since I don't define an executor at all in my file and still prefect runs my flows perfectly fine and according to the docs

it seems that by not setting an environment the default environment is used. The default environment loads the default executor which is the localExecutor which the docs say

An executor that runs all functions synchronously and immediately in the main thread. To be used mainly for debugging purposes.

Howdy Prefect!

hey y'all, do tasks support returning generators? Currently, when I return a generator from a task, the results don't get returned or get picked up in the next task. Couldn't find anything in the docs about it

Hey, I'm reading about state handlers and I understand how to use them to send a slack notification if a task fails. I'm wondering how I can do something like: task 1 - Check if network connection is healthy task 1.1 - If network connection is healthy then go do the rest of the tasks task 1.2 - If network connection is not healthy then reset connection and return to task 1 task 2 - Something task 3 - Something task 4 - Something The issue I'm trying to workaround is I have a task running on a raspberry pi, and the network keeps giving out, and I don't want to blindly reset it each time, rather I'd prefer to check if its on, then reset it if needed. I saw tasks can be triggered based on "all_failed", "any_failed" etc. But I guess in this scenario I only care if the one that directly precedes it failed or not.

Hello Prefecters! Wondering how one might use the resource manager (https://docs.prefect.io/api/latest/tasks/resources.html#functions) with the imperative API for flows (https://docs.prefect.io/core/concepts/flows.html#imperative-api). I'm using the imperative API with a for loop to add a list of tasks to the flow. But I would like to pass the result (a SQL connection) from

ResourceManager().setup()

to my tasks and also have cleanup. Thanks!

Hey, i’m new to using prefect - but i’ve been trying to follow along with this tutorial: https://towardsdatascience.com/distributed-data-pipelines-made-easy-with-aws-eks-and-prefect-106984923b30 I have most of it working, pods get spun up when i run a flow. However, they are unable to pull the image from ECR that I have pushed. I keep getting the error (i have replaced the account number and repo name)

Message: Failed to pull image "<http://z.dkr.ecr.us-east-1.amazonaws.com/r:latest|z.dkr.ecr.us-east-1.amazonaws.com/r:latest>": rpc error: code = NotFound desc = failed to pull and unpack image "<http://z.dkr.ecr.us-east-1.amazonaws.com/r:latest|z.dkr.ecr.us-east-1.amazonaws.com/r:latest>": failed to resolve reference "<http://z.dkr.ecr.us-east-1.amazonaws.com/r:latest|z.dkr.ecr.us-east-1.amazonaws.com/r:latest>": <http://z.dkr.ecr.us-east-1.amazonaws.com/r:latest|z.dkr.ecr.us-east-1.amazonaws.com/r:latest>: not found

I have followed the tutorial pretty closely, although some stuff what slightly outdated so I used the newer paradigms that were introduced. Any insight on how I could configure my EKS to be able to pull images from the ECR would be appreciated. I also did all of this on my master admin account, so I would think that permissions shouldn’t be an issue

Hi, I have a problem were logs for a task run doesn't show up on the in the logs tab but only under the overview (see pics). It doesn't help to try and bring the from "glacier storage". Has anyone tried something similar? I'm using Prefect Cloud.

Hi everyone ! Relatively new to Prefect, we're currently using this (great!) tool to orchestrate our ELT process (with Airbyte & DBT). Until yesterday and since a few weeks, everything was working great but this night, an odd error occurred on my `AirbyteConnectionTask`: •

JOB XXX failed

•

Final state : Success

All my tasks were set as reference, so I really don't know how to explain why this flow showed no trace of error (taskbar was green, etc.). Do you have any clues to explain this behaviour?

Hey all, I have a couple of Prefect flows - both local and Vertex runs. All of them are using docker storage. Now the flows work without any problem - but sometimes I do get the following error:

. 23 January 2022 11:00pm]: 500 Server Error for <http+docker://localhost/v1.41/images/create?tag=2022-01-19t11-58-16-139521-00-00&fromImage=uxxx-flow>: Internal Server Error ("Get "<https://xxx>": context deadline exceeded")

Seems like the docker registry is unavailable for some reason for a short period of time - exactly when Prefect is trying to pull a flow image and run it. I have 2 questions that I couldn’t find any answers to: 1. Is there any configuration option I can use in order to make Prefect retry the image pulling on failure? 2. Can I configure Prefect to pull images only upon change (new flow version) and saving them locally? The way it works right now is that each run invokes an image pull. Thanks!

I get this error, when i try to restart a failed task. How to fix this ?

In Prefect 2.0, Is there any option via config to change the default port number 4200?

Hi All, I have this ELT scenario and appreciate that you could give me suggestions. 1: ET runs as first step that pulls data from a SQL server, saves data into zipped csv files and then uploads them into snowflake by PowerShell scripts. 2: After EL is done, T runs through a dbt project 3: Then we need to process tabular cubes and Power BI datasets through PowerShell scripts and API. I'm thinking a dedicated image for each step and then use prefect to create containers in a workflow, run each step as a task insider this container one by one. and then shutdown containers in the flow's last step. Is that possible? Another way is that creating flow for each step and another "parent" flow to mange the workflow.

Hi all! We have a staging table in Snowflake where we want to update the data validation status of files. We are using Great_expectations to handle the validations and we see the results under the artifact in the UI. If the validation fails, the flow fails and the log shows failed in the UI. 1) Is there a way to get the status of a flow (FAILED or SUCCESS) in a prefect task so we can update the Snowflake table from the task? 2) Is there a way to get the great_expectations result in a prefect task? Thanks!

Is there a way to restart a flow run from the beginning rather than from failed/cancelled task state?

Hi there. I am getting this error in the

wait_for_flow_run

task when trying to implement a flow of flows:

Unexpected error: TypeError('Object of type FlowRunView is not JSON serializable')

Any ideas? Code in thread.

Is it possible to build the docker image ourselves that prefect uses to execute the code? For instance I have:

flow.storage = Docker(registry_url="<http://x.dkr.ecr.us-east-1.amazonaws.com|x.dkr.ecr.us-east-1.amazonaws.com>",
                         image_name="y",
                         image_tag='latest')
flow.executor = DaskExecutor()
flow.run_config = KubernetesRun(env={})

# flow.run()
flow.register(project_name="SA-Test", build=False)

I want to build this manually as part of our CI/CD and push it to the ECR. That part is done. Now when I want to run the prefect flow, the kub pod gets the image but then I get an error:

Failed to load and execute Flow's environment: ValueError('Flow is not contained in this Storage',)

Is there anything special I need to do in my Dockerfile or when I build this image to allow it to be successfully used without building inside Prefect

My team is using Git storage with prefect cloud. We plan to have 2 projects: 1 for dev and 1 for prod. We would like the flows in the dev project to all pull from a “dev” branch of their respective git repos, and the flows in the prod project to pull from a “prod” branch. What’s the standard pattern for doing this? My understanding is that the storage is defined when we register the flow to prefect cloud. Is this correct? If so, is there a way to parameterize the branch_name argument when we use when we initialize Git storage? Thanks in advance for any help!

Hi Community, I have a Flow configured to take in a parameter and after a few processing on the parameter, It sends it to a mapped task for further processing. Problem is, to be able to debug it locally on my computer, I’m facing a few issues. I have been following the debugging guides and not sure what is it that I’m doing wrong.

with Flow("Reporting-Data-Import",
          storage=Local(path="/app/flow.py", stored_as_script=True),
          run_config=DockerRun(image="reporting_data_import:prod-prefect")) as flow:
    # load parameters
    with open("parameter.json", "r") as f:
        default_param = f.read()
    # 
    json_param = Parameter("data_sources", default=default_param if default_param else "").default
    flow_parameter = extract_parameters(json_param)
    is_valid = is_parameter_valid(flow_parameter)
    with case(is_valid, False):
        <http://logger.info|logger.info>(f"Invalid parameter! {flow_parameter}!")
    # copy table & content
    copy_table_content.map(param=flow_parameter.sources, upstream_tasks=[is_valid])

Please share your thoughts as I’m not sure where my issue is.

Hi,Below is my state handler snippet, i am getting error

'Running' object has no attribute 'is_success'

,can someone help.

def post_to_cw(obj, old_state, new_state):
    if new_state.is_failed():
        cloudwatch.put_metrics(cid=CID, flow_name=FLOW_NAME, status_code=1)
    if new_state.is_success():
        cloudwatch.put_metrics(cid=CID, flow_name=FLOW_NAME, status_code=0)
    return new_state

I have what may be a dumb question. I've set my tasks to have multiple retries. I've noticed that if the first try fails, the others will always fail as well, no matter how long the delay. Example - I lose connection to our database during an extract task. 2 minutes pass before a retry, and it loses the connection again. A day later (next run), things go fine. So, are the retries actually doing anything?

Hi again. Is there an idiom to run mapped tasks respecting the order of the item in the sequence? I believe I can use

LocalExecutor

and it will run them in order but was wondering if there is a way to explicitly define the dependencies. For example, in this case, I'd like to make sure that the mapped task for 2 runs after 1 and 3 runs after 3.

items = [1, 2, 3]
ny_task.map(items)

thanks

What is the best way to use a single client object across some mapped tasks?

Hello. I'm looking for some information on how to call ecs Task Defintions from within a Flow. I've been pointed to AWSClientWait as a method, but I can't find any illustrative examples on how to kick it off. Any help or guidance would be greatly appreciated.

The child task is stuck in pending state even if its upstream tasks are all successful. How to fix this ?

Hello! I'm having a little bug with automation, a few days ago I was struggling with emails notifications when sending to an alias, so I switched to slack notifications We realized it was due to restrictions on the alias (only mails from our organization were allowed) We removed this restriction and now prefect sends us notifications by email, BUT I deleted all the email automation and the mail keeps coming, what should i do ?

hi all, trying to schedule ML predictions using prefect. this involves loading a pre trained pickled ML model, refitting, and saving new model daily. The very first model can be saved either using a pickle/joblib backend I know how to write manually the code for reading/writing the model locally/to s3, but has anyone done this using

LocalResult

and

S3Result

?

Hi all 👋 I'm a little confused regarding using regular functions (not tasks) in flows and cannot find information about this in the docs. Maybe I did not look in the write place ? For instance, in

with Flow("my-flow") as f:
    a = make_a()
    b = make_b()
    c  = a + b

• Do

make_a

and

make_b

need to be tasks or can they be regular functions ? • In particular, when using builtin functions (for instance,

datetime.utcnow()

), can we use it directly or must we wrap it in a task ? • Is

c = a + b

allowed ? Thanks a lot !

Can context secrets be passed from agent to dask executors during flow run? Sorry if im repeating this question. Im trying to get some credentials available on the dask workers, I have set envs like

PREFECT__CONTEXT__SECRETS__DB_URI

on the agent. Is this supposed to be available as prefect.context.secrets["DB_URI"] when the flow is run on dask worker without any other setup?