Hello! Is that mandatory to create an agent when running the flow of flow? I have been reading this example. But any agents there If created one why it is running too many processes? Should agent be killed after each run in scheduled task or is it done by program?. Otherwise it is eating too much RAM and increasing each time it runs. how to kill the agent after process.

Getting the following traceback while shutting down the orion server.(running locally) Did not get such errors a few days ago. Confused what is this stack trace trying to say

Prefect 2.0 - Snowflake PUT command from local file error

Hey All, Apologies if this is the wrong channel for this. We've come into a bit of an issue with creating a dynamic flow in Prefect 1.0, see below:

with Flow("tax_assessor_inc") as flow:
    
    raw_s3_keys = Parameter("raw_s3_keys")
    
    flow_tasks = []
    for s3_key in raw_s3_keys:
        _clean = clean([s3_key])
        _load_data = load_data(_clean)
        _test_raw_data = test_raw_data()
        _transform_task = transform_task()
        flow_tasks.append([_clean, _load_data, _test_raw_data, _transform_task])

    reduced_flow_tasks = sum(flow_tasks, [])
    for i in range(len(reduced_flow_tasks) - 1):
        reduced_flow_tasks[i].set_downstream(reduced_flow_tasks[i + 1])

The issue arises when trying to iterate over the

raw_s3_keys

parameter as it's not iterable. It works when hardcoding the s3 paths but obviously that's not what we want, as it would great to dynamically define that list. Also annoyingly each task has to occur synchronously otherwise we'd use something like prefect map. Firstly is it possible to iterate the output of a parameter object. If not is there a better way of solving this problem. Sorry for the long winded message, Thank you 🙂

What are my options for alerting with Prefect 2.0? I guess I can create alert tasks and then call them as part of a flow on failure. Is there any plan or ability to handle that on a higher level? Such as on Cloud?

Good afternoon 🙂 I am on Prefect 1.0 and was asking myself - mainly out of curiosity - if it is possible to exclude a specific element while calling .map(), store that element and process it at a later point within the flow? (Not talking about unmapped)

Hi there, we are using an ECS agent to run our flows. is it possible to set the cpu and memory per flow. I tried setting the cpu and memory for the flow in ECSRun config but the task that gets spun up still uses the default that was specified in the yaml file i.e cpu 1024 and memory 2048. i want to change the cpu and memory per flow. my current ecs run config is as follows:

RUN_CONFIG = ECSRun(
    labels=["dev"],
    task_role_arn="arn:aws:iam::xxxxx:role/prefectTaskRole",
    execution_role_arn="arn:aws:iam::xxxxx:role/prefectECSAgentTaskExecutionRole",
    image="<http://xxxxxx.dkr.ecr.us-east-1.amazonaws.com/prefect-orchestration:latest|xxxxxx.dkr.ecr.us-east-1.amazonaws.com/prefect-orchestration:latest>",      # this is our custom image that also has our flows
    run_task_kwargs=dict(cluster="prefectEcsClusterDev", launchType="FARGATE", overrides=dict(
        containerOverrides=[
            dict(
                name="flow",
                cpu=4096,
                memory=8192,
            )])),
    cpu=4096,                    #tried specifying here but still no luck
    memory=8192
)

Prefect 2.0 Add users & edit permissions

Hello, I’m not seeing logs on our flow runs, have a feeling that it may be something on our account — is there a way to double check?

Hi 🙂 wondering if anyone has resources for creating a (near) real-time data pipeline using Fivetran and Prefect?

Question about environment variables, if I set something like

"FOO": "BAR"

in the run config, and then do something like

os.environ["FOO"]: "BAZ"

in the flow.py - what will

FOO

bet set to when the flow runs?

Hi, today I setup a personal account for development using personal email address. Projects and flows are registered under my account. For production deployment, we want multiple people to be able to monitor flow run and access Prefect Cloud UI. What is best practice on setting this up? Should we setup a service account (share email address) and everyone login to same account?

Hello all! Are there any release notes for the 2.0b5 release anywhere?

Hi all! I am currently trying to use Prefect together with Ray using the @flow(task_runner=RayTaskRunner()) annotation. While on the head node of the cluster everything works fine, on the workers the flows continously fail with the message: "(begin_task_run pid=1659, ip=172.31.2.206) File "/home/ray/anaconda3/lib/python3.7/site-packages/prefect/orion/models/task_runs.py", line 282, in set_task_run_state (begin_task_run pid=1659, ip=172.31.2.206) raise ValueError(f"Invalid task run: {task_run_id}") (begin_task_run pid=1659, ip=172.31.2.206) ValueError: Invalid task run: 3e46e4ce-c740-47d8-9ac7-79951b6fd98f". Any idea why? I am using ray 1.12.1, prefect 2.0b5 and python 3.7.7.

Hi, I was trying to find some information on using redis as a store to persist data between task runs (or flow runs). Is there any? Thx!

Hi , What can be the reasons for flowstorage error while running a flow in prefect cloud

Hello 🙂 We're having issues with tasks that require manual validation: they won't start 😕

In Prefect 1.0, is there a way to run the flow next on schedule only if its earlier run is successful? Looking for program based solutions but other also welcome! Thanks.

Will decorating class constructors as task and a function that pipe those tasks as flow work with prefect 2.0 In cases where we create deployments What is your opinion and what do you suggest? Should a function be created where that constructor is called and decorating that with flow works better than the previous approach???

Hi! This morning i was refactoring some code and also upgraded to prefect2.0b5 - but when i tried to deploy and run my agent on a “Compute Engine” instance i ran into problems. (Dont think it’s related to prefect2.0b5 since the problem can be reproduced in b4 as well. When i run a flow from the compute engine i get an exception saying that the client cannot be pickled:

_pickle.PicklingError: Pickling client objects is explicitly not supported.
Clients have non-trivial state that is local and unpickleable.

After reading up on this it sounds like this is due to some other exception that results in an unpickleable object. Is that right? Are there any ways i can reach the underlying exception? Traceback and code in thread..

Hi, we’ve got a flow of flows, which runs every time a file is added and then at the end runs a flow that updates a dataset in Power BI. Since we’re only able to refresh PowerBI 8 times a day, is there a way to group the power bi refreshes together in case multiple files are added? Just wondering what might be the best way to architect that, or if anyone’s done something similar

Hi everyone - is there an easy way to get the Docker container id for a flow run? We are currently get it by checking the prefect logs for the flow run, but I’m wondering if there’s a better way. We’re using the Docker agent and we are on prefect 0.15.13.

Hey everyone! I am currently evaluating the use of Prefect Cloud as the main workflow automation tool for a client. I have been trying to find details on where exactly the servers would be hosted, but information seems scarce on the website on that topic. Because of EU privacy laws, guaranteeing that our data never leaves the EU is a must-have requirement. Did I miss something on the Prefect website, or is there maybe someone here with similar issues? Thanks a lot 🙂

Is there a way to set the start time of a IntervalClock for a flow using an input parameter?

Hey, i finally deployed prefect on a k8's cluster on aws thanks to your advices ! I was wondering if i could use the

local dask executor

in this cluster or if there is an aditionnal configuration to use dask as an executor in a k8's cluster ?

wha is the default cpu_request for a kubernetes run_config?

I have a flow with a section of mapped tasks, one of them failed due to a missing heartbeat - the final flow state is

success

though. why wouldn't this flow's overall state be set to

failed

?

Hi everyone! P Just FYI: I'm OOO for the rest of the week - if you have any urgent requests, Kevin is around. I'll respond to any open threads you have with me next week. If any of this resolves itself until then, or if you have some updates, it would be great if you could add a short message about the current state. Thanks and have a great rest of the week! 🙌

Is there any reason the scheduled jobs can't run on-time if it is scheduled one time using RRule? I see it is scheduled with correct DTSTART and Count = 1. But prefect agent is not picking up and running it immediately. Prefect Orion 2.0b

Is there any reason why flow runs would suddenly not be able to be scheduled? I didn't make any updates at all to my server or agent and since Thursday I'm not able to run any flows. I'm using a kubernetes agent. I have no idea why this started happening or how to go about fixing this