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 👋 @Zanie 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!

@Zanie 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?

I've got an

ECSAgent

running (in ECS, using the base prefect image with

prefect agent ecs start

), but the tasks it spins up are in the wrong subnet so I want to customise the task definition. It looks like I can pass in a task definition path but that's a bit troublesome because now I need to store file somewhere accessible to the agent. Is there a way to pass a task definition ARN instead? It looks like I can pass that to

ECSRun

but the subnet ID is generated by Terraform so I can't figure out how to get it into the Python...

I'm also not totally clear on the relationship between ecsagent and ecsrun. If I didn't use ecsrun, would the flow run on the ECS service that's running the agent directly?

Hi Prefect Community. I’ve been trialing Prefect to replace Airflow and documented my thoughts to date here, along with a brief setup guide, not sure if this would be useful for others: https://jamesutton.medium.com/my-experience-with-prefect-so-far-and-a-quick-getting-sta[…]bc?source=friends_link&sk=58e6e074ee852ffbef2e46999d43b1c7

Hi! Im trying to start a run though gql. I am using this mutation where the input is a variable: Anyone know what the type is supposed to be for the input?

mutation MyMutation($input: any) {
    create_flow_run(
      input: $input
    ) {
      id
    }
  }

Hi I need to add Prefect Cloud API IPs for whitelisting in Security group of EC2 where Agent is running ... Couldn't find IP range... Anyone know where can I find it?

Question about Orion: generic tasks, flow build without running it. Details in thread.

I'm trying out Prefect. Two questions: • I added and registered a flow with a local server. It has caught my attention that when I re-add the exact same flow, a new 'version' is created. What is the best practice for adding a flow? Am I supposed to add it just once and again for updates? Or always before creating a new flor run? • What is the typical separation of concerns? Should I just call

create_flow_run

in the client, or should the client be doing anything else?

Hi, It is possible to fix MAC address when docker storage is using ? Docker run command have

docker run --mac-address="70:ca:9b:ce:67:ae" IMAGE

So I want set constant mac address when define docker storage as:

storage = Docker(
    env_vars={"PYTHONPATH": "$PYTHONPATH:/pipeline"},
    files={f'{parent_dir}': '/pipeline'},
    image_tag=os.environ.get('IMAGE_TAG'),
    image_name=flow_name,
    stored_as_script=True,
    path='/pipeline/flow.py',
    extra_dockerfile_commands=[]
)

Someone has tried to achieve something similar ?