Does anyone have agent deployments in multiple cloud providers pointing to the same prefect cloud account? I have a use case where I plan to access some data in GCP, while my agents are deployed on AWS EKS clusters - I’ve considered going with some sort of peering approach, or simply standing up an agent on GCP GKE

Hi, do you may know how to add Context from the UI? I have variable defined in

[context]

local

/.prefect/config.yaml

but when released to Cloud, it says the CONTEXT not exists. But I don’t find where can create that on UI.

Hi, I am looking for a Prefect task for doing an BigQuery export to GCS. Something similar to BigQueryTask but instead of writing to a

table_dest

, to write to a GCS bucket. Is there a Task or a way to do this?

I'm running the following flow using

create_flow_run

to run a flow multiple times with different parameters. Here is my code and the error i'm getting.

Hi How to call outside flow from task. I am getting error

is it possible to get prefect to take in a boolean parameter? Or do all parameters need to be strings?

Hello, It might be a silly question but I can't really figure out the answer. One of the tasks of our flow is retried while it was previously already successful (see screenshot). And because this is a task calling a DELETE API endpoint, the retried task failed (because the object has already been removed). Although I could easily handle such thing, I would like to understand a bit more what's going on. I suspect something failed somewhere in the flow, and the flow is restarted from scratch, and all the tasks are restarted? I could not really find anything in the doc about that. What would cause a flow to retry a task that has been previously successful? Some precision: prefect core 0.15.12 prefect server 2022.01.12

Hi Flowrun values is same when i call the task from other task . i want generate multiple flow run for each call to task

Hi, I have a Flow_1 which runs on a Agent_A (machine A), after completing few task, Flow_1's task has start/trigger another flow Flow_2's task on another Agent_B (machine B). Each Code of Flow_1 and Flow_2 are present on different machines(Agent A and B). Basically Flow_2 has tasks whose upstream task are present in Flow_1. I have read about flow_to_flow in prefect but that works for 1 Agent and cannot be mapped for Flow_1.task_A => Flow_2.task_B . How can i achieve this ?

@Kevin Kho When using the

KubernetesRun

run configuration, I want to specify a custom Kubernetes Job spec in order to update the

envFrom:

value. Is there an appropriate template spec that I can use as a starting point?

Hello! I'm trying to figure out the way to organize code in my flow. I'm using Kubernetes agent and GitLab storage so my flow code is being pulled from GitLab into kubernetes pod. The problem is that while Prefect clones the whole repo, other python modules from the same repo are not available for import. It is not an option to store the whole flow code in a single file as it can get enormously large. How do you handle code organization within your flows?

Hello, I have a task that I am calling map on using Dask executor which works fine until there is an error. Sometimes I get an "Unexpected error: KilledWorker" which in my case seems to be because of dask worker has run out of memory. There are only a handful of cases where this happens but when it does it kills the whole flow run. Is there any way to handle this without killing the whole flow? I want to have 2 downstream tasks that would take the filtered results and act on them. Any help would be appreciated.

Hi! When writing DWH tables from Prefect, are there any recommended metadata fields that should be attached to the rows. I'm thinking about things like flow label and version as well as created / updated timestamps. Are there ready to use functions in Prefect for this?

Is there an example of a repo or something that has a nice set up for production grade Prefect code (particularly flows, tasks, etc)? We use Prefect on my team but right now somewhat struggling to find a good set up for re-usable tasks and flows that are well tested. We also use prefect server at the moment self hosted on K8s

Hi all, I'm running into a issue with RunGreatExpectationsValidation, hope you can help. When I run it directly

flow.run()

it works fine. When I register the same flow to a local server and run it with a local prefect agent I get the following error.

File "/home/prinsn/Code/prefect/repo/gima-prefect-cicd/.env/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 602, in dump
    return Pickler.dump(self, obj)
TypeError: 'NoneType' object is not callable

I' m using: ge api V3 prefect==0.15.13 great_expectations==0.14.5 Any pointers? =D

Hi all, does anyone here have with running Prefect flows using AWS ECS tasks on EC2? My org currently runs a lot of flows using ECS tasks on Fargate, but Fargate has some constraints that EC2 doesn't (e.g. you can't use GPU-enabled instances w/ Fargate). In order to accommodate some of these requests, I was considering either: 1. Launching a new EC2 instance (that has all the hardware requirements that the requesting teams need) on which we'd have a DockerAgent running, and then use labels to manage which

DockerRun

flows will or won't run on that beefed-up EC2 instance 2. Using ECS on EC2, and then simply manage the execution of the flows on the EC2 instance by configuring an

ECSRun

flow to use an ECS task that executes on EC2 The former option seems a little more straight-forward, and we could get it up and running pretty quickly, but it would involve overhead of managing labels, new API keys for the agents, etc. The latter is probably more flexible in the end, but there's more up front work for us since ECS on EC2 is not a workflow that we currently support. Has anyone ever deliberated over this issue or experimented with it?

Hey all 👋 @Michael Adkins is our primary guest this week on PrefectLive. We're going live in 5 minutes, come hang out in the chat and ask Michael the most complex question you can think of. We also have a special guest joining us - our very own @justabill! See you at twitch.tv/prefectlive!

@Michael Adkins has been leading the effort in making Orion async-first, parallel by default, and more memory efficient so if you attend one Twitch stream this is the one ^^

Hi, I see a weird issue when registering the flow. Do I have to change anything to make it work? (please see exception logs in this thread)

Hi, I have task A and task B, both mapped and A is the upstream task of B. Is there a way of having the successful children task in A move on to B regardless of the state of other children tasks in A?

Hey guys i getting this error while running my flow. Is there something i can look up to verify this error? I'm using the correct images from docker

 Failed to load and execute Flow's environment: StorageError('An error occurred while unpickling the flow:\n  ModuleNotFoundError("No module named \'testing_flow\'")\nThis may be due to one of the following version mismatches between the flow build and execution environments:\n  - cloudpickle: (flow built with \'2.0.0\', currently running with \'1.6.0\')\n  - python: (flow built with \'3.9.7\', currently running with \'3.7.10\')\nThis also may be due to a missing Python module in your current environment. Please ensure you have all required flow dependencies installed.')

Hi, I’m using

GITLAB

as storage and

Kubernetes

as agent to run prefect flows. I noticed I get dependency errors when a custom pip package is used inside the flow, in this case soda:

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named 'sodasql'")

Is there a way to address these dependency issues without using Docker images to store the flows?

I’m starting to use

create_flow_run

,

wait_for_flow_run

, and

map

to fan-out via sub-flows. Is there any way to get more friendly task names using these tasks so I can more easily identify what child task is still running? Right now I have to go into the parent flow, look under logs for the URL for the created child flow run and then navigate separately. Ideally (maybe in Orion some day) it would be nice if the lineage between parent and child flows could be visualized; however, I realize this may be difficult thing to do. I’m just trying to find ways to make it as user friendly for an engineer to go in and understand which sub-flows failed or are running/etc.

Does "storage" mean the storage used to store the definitions of flows? The caching of task state? Eventual results? If it's more than one of the above, in particular if it's both the eventual results store and the flow definition store, then how can I separate those?

I'm trying to get prefect running with Dask, using Prefect Cloud. I'm assuming here that Prefect Cloud is whatever runs or helps running flows that are shown in the prefect.io UI. I have some questions about terminology, which could mostly be summarized by asking "What exactly do flow runners, executors, agents do? What is the timeline of their communication with each other? Where do they run?" FlowRunners handle the execution of Flows and determine the State of a Flow before, during and after the Flow is run. A flow's Executor is responsible for running tasks in a flow. During execution of a flow run, a flow's executor will be initialized, used to execute all tasks in the flow, then shutdown. Agents orchestrate flow runs. Agents start and monitor flow runs. During operation the agent process queries the Prefect API for any scheduled flow runs, and allocates resources for them on their respective deployment platforms. So, when I call flow.run() with a flow that’s configured to run using Prefect cloud resources and coiled/dask, where is the FlowRunner? Is it my local machine, the process that called flow.run()? Or the cli, with which I launched a flow? In Prefect Cloud, there will be a flow visible on the UI, waiting for an agent with the right flags to connect. What do you call the process that is waiting for an agent to connect? If it’s the flow runner, then what do I call the local process that called flow.run()? If an agent does connect to Prefect Cloud and starts orchestrating the flow, what does it launch? An executor? A flow runner? Where does it run them? How does the flow runner learn the location of the executor? According to documentation, the flow runner loops over tasks and sends them to an executor. I’m guessing the executor then finds resources on which to launch TaskRunners. If it’s a DaskExecutor, I’m guessing it looks for Dask workers I’ve provided. However, it seems as though it does not launch dask workers, because that’s what an Agent does. If Agents allocate resources, and Dask is a recommended source of resources, why is there no DaskAgent? Why only Docker, Kubernetes, Vertex, and ECS?

Hi Guys, My name is Marcos Lopes, I’m from Brazil and I have been working on Operations Director at IntuitiveCare. I love programming and I’m studying now Prefect to help me and my team to create some automations process for IC.

Hi. How is the logic around retries and resummiting of flows? I have a flow with some short retry delays (30 secs). This flow will stay running and wait for retry. I have another flow were the retry delay is longer (15 min) and here the flow gets resubmitted. What is the difference here? It yields some issues as there is no need for storing results in first scenario (they stay in memory) while in the second scenario they must stored so they can be used when the flow run is re-run.

I need some help understanding memory limits in

DaskExecutor

! If I run the code below without the

DaskExecutor

, it gets to 40/50GB before being Killed because of OOM. However, if I run it as-is (i.e., on

DaskExecutor

), it fails at 10GB, tries three times, and then gives a

distributed.scheduler.KilledWorker

. Does Prefect do anything to the Dask worker memory limits? I'm facing this issue running with KubernetesRun and a DaskExecutor with a KubeCluster backend, where the worker spec has 50GB of memory for both k8s and the dask worker limit. It still fails around the 10GB mark. (In practise this is while loading PyTorch models and predicting.)

import numpy as np
import prefect
from prefect.executors import DaskExecutor
from prefect import Flow, task

@task
def test_memory_limits():
    logger = prefect.context.get("logger")
    for size in [1, 2, 5, 10, 20, 30, 40, 50]:
        logger.warning(f"Creating array with {size=} GB")
        a = np.ones((size * 1000, 1000, 1000), dtype=np.uint8)
        logger.warning(f"Created with size {a.nbytes/1e9=} GB")
        del a
        logger.warning("Deleted")


with Flow(
    "test-memory",
    executor=DaskExecutor(),
) as flow:
    _ = test_memory_limits()

does anyone have good patterns for registering flows via ci/cd? ideally id like to register only the flows that changed automatically on a git push, and be able to use different run configs for each, while also handling env vars that may be different in prod/dev. currently i have a script that loops through all my files in my flows directory and registers all flows using extract_flow_from_file. it runs on a push but the way i have it set up it: • always registers all flows, regardless of whether they have changed or not • has to pass the same env values to all flows, regardless of what the flow actually needs • passes the same run config to all flows, not allowing me to customize memory requests per flow any guidance on what to do here?