hey people, I’m having some issued with a specific use case: I have a large amount (e.g 10000) of objects which I want to perform am operation on, in separate tasks each has overhead of loading external data, starting a pod, etc… so…. I want to perform this operation in chunks (e.g, 100 objects per task). The problem is that passing a mapped array of arrays doesn’t really work for me, and I get this (attached) is there a way to mark the inner lists as

unmapped

? (code added in a reply)

Hello everybody. I have a Flow that create_flow_run (a few) and waits until the finish - wait_for_flow_run. But if one of the sub-flow is Failed parent flow has a Success state anyway. Can I make parent flow Failed if one of the children fails?

Hey ! I’d like to share an issue I’m facing with Prefect that someone that uses Kubernetes might be able to help. I’m registering workflows using Docker storage and the registration + build of the image happens through a CI/CD pipeline that runs on Kubernetes (AWS EKS). The Environment: As you saw above, my CI/CD environment is containerised (a pod on Kubernetes). This means that when registering a flow the prefect register API is building a container inside a container. The issue: The Prefect register command (which builds the image) fails to pull libraries from PyPi throwing connectivity issues (DNS resolution). The Kubernetes pod has internet access + DNS working as I can pull the prefect Library from PyPi for example. But for some reason the registration command where the image is built throws this:

WARNING: Retrying (Retry(total=4, connect=None, read=None, redirect=None, status=None)) after connection broken by 'NewConnectionError('<pip._vendor.urllib3.connection.HTTPSConnection object at 0x7ffb221c2160>: Failed to establish a new connection: [Errno -3] Temporary failure in name resolution')': /simple/pip/

see the

Temporary failure in name resolution

In summary: I’m not sure how much this is Kubernetes or a Prefect issue. I’m throwing it here as there could be a tweak on Prefect we could do to make it work. Any ideas ?

Hello folks, i have a question reg Prefect cloud flow/task execution and IP range of cloud. Here is what my current flow does: • single flow having multiple tasks to sync few S3 buckets. (each task runs s3.sync command via ShellTask ) • catch 22: the source S3 bucket is from different account (lets say X) and destination is bucket is in different account (lets say Y). The AWS policies are already in place to do cross account s3 sync. (the IAM role/user are already added in policy for access) Issue: • I need an IP address/IP range of network where this S3 task execution happens so that we can whitelist the IP range of this at source bucket account (X) policy. I am running my task on Prefect cloud - which means the request to execute sync command goes from prefect cloud network. Is there any way i can get the IP range of prefect cloud where this execution happens ? I need to whitelist the IP range. In case if i run prefect server on my infra then it becomes easy for me to whitelist the IP range of an AWS VPC under which the server instance runs. But in prefect cloud i am not having clarity. cc: @Kevin Kho

Hello Prefect community! I'm working on a machine learning team that hopes to use Prefect to manage ML feature pipelines and model training/ops with Prefect. We're just getting started. If there are any other data science or ml practitioners further in their journey with Prefect, we'd love some advice or "I wish I had known/started with"s 👍

Could I get some suggestions on how to debug a flow which has been submitted for execution, but doesn't actually start? I've confirmed: • the agent (ECS) is querying for runs • the flow is tagged properly for the agent to pick it up • cloudwatch tells me the agent properly picks up the flow

is there anyway to set the

max_retries

on a task from the UI (or without having to upload new code)? I was hoping I could template the value and pass it to the @task decorator, but it doesn't look like you can use templates with the

max_retries

argument

I'm new to Prefect. I've a scenario and I cant seem to find a way in prefect to do it. I've 3 flows - A, B, C all running on their own schedules. Flow B and C have an indirect dependency on A (the latest run of A should be completed and not failed). I was trying to check the status of the latest flow run using FlowRunView (which takes in flow_run_id as parameter) inside Flow B and C. Is there an api to get the latest flow_run_id for a flow (using its name)? Is there any other workaround?

We're a bit behind on Prefect versions (0.13.19) because we've been using Environments (especially DaskCloudProviderEnvironment) so we're ramping up on Run Configs. With Run Configs can we: 1. Examine the parameters of a Flow run at the start of execution, calculate the number of Dask workers based on a parameter value, and create an ephemeral Dask cluster to run that Flow on? 2. Alternatively, associate a Run Config with a schedule, e.g. Run this Flow at midnight with 12 workers and run it again at 4am with 20 workers, etc.

Hi, I'm trying to run a Pyspark script inside my Flow, but it gives me this error: I'm using Local Agent, already tried changing environment paths, but still gives me the same error after I register -> "Quick Run" Does anybody know how to deal with pyspark scripts(not Databricks) with Prefect?

Exception: Java gateway process exited before sending its port number

Is there a canonical way or a canonical task that folks use for writing arbitrary tasks to flow logs? Like if I were to have a (built-in) task whose output I'd like to log somewhere, do people generally just write a trivial task that looks something like:

from prefect.utilities.logging import get_logger
from prefect import task

@task
def log_value(val: str) -> None:
    logger = get_logger()
    <http://logger.info|logger.info>(val)

Hi all, for a prefect flow to pass on credentials to the mapped tasks - can I pass a dictionary of credentials or should be split into singular values ?

with Flow("example") as flow:

  credentials = {'id': ... , 'pass': ...}
  apply_map(transform ,result ,unmapped(credentials)) # can we do this ?
  apply_map(transform ,result ,id =unmapped(credentials['id']), pass =unmapped(credentials['pass'])) # or is this the standard way?

I have an docker image with a specific version of prefect installed that I am using as the base image in my Storage object, however, when i go to registered the flow, a different version of prefect gets installed (i.e. the one I have installed locally where i'm calling flow.register() from). Is there anyway to avoid this? I know you can specify the prefect version when you create the Storage object, but then it is up to the user creating the flow to know what version to specify and i would prefer to just use the version that is already installed on the base image (as this image can be shared across users / flows)

Hey everyone, I am having a small issue with dynamically named runs and automations in prefect cloud: I have a flow with a

state_handler

that uses the

set_flow_run_name

method on the prefect client to rename itself when it moves to the running state. Now, I have set automations in prefect cloud that alert me when a flow run fails. I am now experiencing the following behaviour: If a flow run fails within a few minutes (approx. 5-10 minutes), I see the prefect generated run name (e.g. neat-kudu) in the alerts. If the flow run fails after a longer time (50 minutes), I see my custom flow run name in the alert. Is this expected? Is there anything I can do on my site to fix it?

Hi! I'm having an issue on Prefect Cloud where a flow appears as failed, despite no tasks having returned an error. The weirdest thing is the flow lasted 2 minutes, but one task is still marked as ongoing (blue) with its timer going up. What I don't understand is if there is still an active task, wouldn't the flow also be active past 2 minutes?

I use Bitbucket Storage for my flow. Does anyone knows the local agent will download all file .py (my own function lib) or only the file defined this flow, when i run flow by local agent. thx

Hi there, I’m new to prefect and GraphQL. I’m trying to trigger a flow based on the flow_name. Is there a way to chain the following GraphQL queries or do i need to send the requests consecutively ? Thanks

query GetLatest {
  project(where: {name: {_eq: "project-test"}}) {
    flows(
      where: {name: {_eq: "flow-test"}, archived: {_eq: false}}
      order_by: {version: desc}
      limit: 1
    ) {
      id
    }
  }
}

mutation TriggerLatest {
  create_flow_run(input: {flow_id: "xxx"}) {
    id
  }
}

hi - this is quite a basic question i think, but I haven't found a clear answer so far... Is the recommendation to run Prefect flows against a shared filesystem or is the only persistence between tasks supposed to be explicit use of S3 etc. ? Some hints that a shared filesystem is expected are the filesystem tasks and the multiple mentions of EFS in the (deprecated) Fargate Agent, plus I see that according to AWS themselves, ECS (which Prefect now uses instead of raw Fargate), is intended to have EFS plugged in. However there are a few mentions of files not being persisted between tasks, including this Dask page. In general it seems that a lot of tasks will be a bit painful/impossible to use if you don't have a shared filesystem. Unless you want to pass large files through Prefect Task return values. Thanks!

Hi there, I am trying to run a set of 3 dependent flows meant to be run sequentially (A->B->C). Is there a way I can have both a schedule for this flow (as stated here) and also have a manual trigger to the said parent flow ? What options do I have to manually trigger the 'parent-flow' apart from prefectUI and GraphQL ?

Hi, is there a way to use a dictionary returned from a task in the with Flow block? i read in some threads that it could be done using an intermediate task to unpack it, any examples for such task?

In order to connect an on-prem hadoop cluster to prefect cloud, I tried to add the prefect API host to the file /etc/hosts. ## prefect API 34.96.72.220 api.prefect.io PING appears to work, while telnet does not. PING api.prefect.io (34.96.72.220) 56(84) bytes of data. telnet api.prefect.io Trying 34.96.72.220... telnet: connect to address 34.96.72.220: Connection timed out Can anyone provide any tips on how to allow communication between the hadoop node and api.prefect.io?

Hi, I was wondering if anyone has any great ideas for how to deal with cancellation of long running queries started from a Prefect task. In my case I’m dealing with PostgreSQL and while psycopg2 will try to gently stop a query when you close a connection, in many cases the reason we are cancelling a flow/task is because that query has been running for too long and is unlikely to finish naturally in an acceptable timeframe. Instead of having to manually kill the query with

pg_terminate_backend

it would be nice if the task getting cancelled would be able to call

pg_terminate_backend

.

Also I had a look at the repos and couldn’t find this feature request, but we rely heavily on parameterized flow runs and it would be great to be able to filter runs in the Cloud dashboards by parameters used.

hey, so it looks like a dependent flow run has just disappeared… it’s not archived, just gone. 😕 it was a dependent flow from a flow of flows. the link to the flow run doesn’t load anything anymore, it just brings up a blank page. it’s also not in the list of runs for the flow, nor in the dashboard with the history of every flow run.

Hello, is there any way to stop flows from running on certain dates?

Hi, I use python threading package in a Prefect task, the target functions works properly, but the context logger inside the target function doesn’t show in UI Logs, is there any way to catch the log in Prefect?

Hello, I have a mapped task B, in which, depending on the data obtained, I either generate the data (artifact) or not. This data is passed to the next mapped task C. If I do not generate data in task B, then I want to Skip it, and task C. How easy way for me to do it?

hello! I can't find in the documentation, how can I start an agent, that will run flows without labels?

Hi again, After getting some great help here, I have two new things I am struggling with, perhaps someone here has some input. Challenge 1 Firstly, we have a pretty common use case where we have a flow that receives a list of inputs

[x1,x2,...xn]

that can be processed in parallel through a DAG. Out of this

n

inputs, let's say that a fraction of those

k

fail due to various reasons. We would then find the issues, fix them and like to rerun the flow for those inputs. When

n

and

k

is large, then it is quite some work to go through the UI and/or logs to find all the inputs that fail in order to start the next run. What I am doing right now is naming the tasks with their inputs, that way I can filter failed tasks in the UI. This is not optimal, as it is still involves quite some manual work copying and pasting stuff. I initially thought about logging which inputs failed at the end of the flow execution. However, I cannot think of a trivial way of doing this without it involving some form of global variable that needs to be passed around to different tasks and be write/read thread safe. It just does not sound like the proper way of tackling this. How do others approach this? Challenge 2 Second, my tasks are spinning up Kubernetes jobs/pods to run computational heavy workloads. If the task execution goes well, then I use

DeleteNameSpaceJob

to clean up once a job completes. But when I manually cancel those flow runs or once a job fails, I end up with lots of clutter that I need to manually clean up in the k8s cluster. I have been thinking about using state handlers to deal with this, calling the

DeleteNameSpaceJob

once the new state is

Cancelled

or

Failed

. However, the state handler needs to know the job name it needs to delete, which is a piece of information that the tasks knows about. My struggle is on how to send the

job_name

from the task to the state handler? As I understand the state handler has a specific signature, so I cannot pass extra parameters. I tried using the context for that, but what I have written in the context within the task does not seem to be propagated to the state handler. Any ideas here? Thanks!

Does anyone have tried run a pyspark ( spark cluster in local docker ) task from a local agent in Docker without using bash command ?