In prefect 1.0 I could map() a task and then downstream I could process the list that resulted from that mapping of tasks. In Prefect 2, I can approximate the parallel mapping of tasks using

.submit()

, but I cannot find a way to reduce the results into a format that can be used by a non-mapped/non-async task. I need to find the equivalent in Prefect 2, preferrably using ConcurrentTaskRunner rather than asyncio. It seems like it would be a common use case and I'm not sure why it is not available in Prefect 2.

Is there any way to send a notification [automation], when 2 of the latest run of a single flow fails? In the documentation for automation I see

threshold

and

within

parameters on Automation API, but different flows may have different schedules hence the

within

parameter is not enough. ++ Making a automation for each flow with different

within

parameter value doesnt sound fun esp. when the rule is pretty simple.

Hi folks, I've built my first Prefect flow—it interacts with two banking APIs—and I'm ready to deploy it to production, but I'm struggling a bit to wrap my head around all the pieces I now seem to need.. I figured I'd deploy it to Prefect Cloud for simplicity's sake, but (a) my custom block doesn't seem to be working (it says it 'failed validation'), and (b) the docs seem to suggest that even though I'm running it in Prefect Cloud, I still need to bring my own storage (e.g. S3 bucket) and infrastructure (e.g. k8s cluster)? Am I understanding that correctly, or is there a simpler way to get a basic flow running in production? (Regarding (a), I created a custom block in a local package that just stores client_id/client_secret and has a refresh_token function. I import it via

from prefect_monzo import MonzoCredentials

)

hey! I'm trying to take 1 dbt cloud job, once complete it kicks off two paths of another dbt job run and a set of hightouch jobs. I can get this to run sequentially however I would like the dbt job and hightouch job to run concurrently. When i try async it gives me errors about the credentials not being valid and I noticed it send

coroutine

instead of the credential dict. Wondering if anyone has successfully pulled off what I am attempting. I assume its not working because I am taking prebuilt flows but not sure. Thanks for the insight!

Hello everyone! I am facing some issues in our deployment after the owner changed in our instance.. Right now I'm seeing this error message from the agent in the compute engine in GCP: "httpx.HTTPStatusError: Client error '401 Unauthorized' for url 'https://api.prefect.cloud/api/accounts/8cba8c22-3359-43ea-845e-b88737a33ddd/workspaces/caa7e1bf-faa2-4ba5-baab-1987efa226f1/work_queues/name/prod' For more information check: https://httpstatuses.com/401 An exception occurred." Getting this when checking the /health endpoint (see image) However, I've updated the PREFECT_API_URL and the PREFECT_API_KEY. Locally flows run fine. Not sure what's causing this and how to fix it.. would anyone be able to point me in the right direction?

Hi guys, I am trying to set my prefect flow to use temporary dask clusters in GCP but it fails most of the time and the latest error is

prefect.exceptions.ClientError: [{'path': ['set_task_run_states'], 'message': 'State update failed for task run ID e18b41ed-7ce8-449f-bda2-6de3956eb555: provided a running state but associated flow run 8484f5dc-acef-4ac5-ae08-9bf6d72edcf2 is not in a running state.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

cloud you guys help me understand what causes this issue?

Is there any way to launch a flow run on k8 and specify the pod name? With the randomly generated names, when we have alerts go off that pods run OOM for example, we have to go through extra steps to trace that back to a specific deployment since the pod name conveys no context.

hey all, seeing a weird issue with tasks not retrying in prefect 1.0 the last couple days. normally we have a task that retries every 30m for a few hours (it depends on a sheet that's manually pushed to sharepoint by an analyst). the flow has been working fine for the last few months, but today and yesterday it hung after the first failure until someone manually kicked it off

Hi to everyone, we are migrating from Prefect 1 to Prefect 2. For storage we are using S3 buckets. We were able to store the flows in S3, but when we usa a local agent for debugging it, it seems like it's looking for the files locally instead of getting them from S3. Also, we are getting this error that may be related (but the flow run doesn't stop at this error):

Flow could not be retrieved from deployment.

I've been testing changing the configuration of the deployment but I had no luck. Permissions to the bucket seem to be ok. do you have any idea of what am I missing? Thanks!

Hi, all! Trying to use CronSchedule from the prefect.server.schemas.schedules library, but get this on deploy: ERROR: Could not find a version that satisfies the requirement prefect.server (from versions: none)

Hi all - we are migrating from Prefect 1 to Prefect 2. 1. We used Projects to organise flows in Prefect 1. Are workspaces equivalent of that in Prefect 2? What is the best way to organise Flows/Deployments? 2. Currently the agent’s (which is on AWS ECS) PREFECT_API_URL is pointing to a specific Workspace. So do I need a separate agent for each of the Workspaces I have? 3. Are there any references to documentation / example code re setting up deployments and Flows in Prefect 2? Thank you!

I have a Prefect deployment that runs every 2 minutes to update data on live sporting events. Two minutes is enough time to run the ETL on an event before the next flow run, but ideally I would like to have a process that is always running and updating an event or multiple events that can be taking place at the same time, and after any update just check for live events again and update as needed. Essentially updating a scoreboard of live games, but without any sort of push from the primary data source. What would best practice be here? Schedule the deployment to run once a day, keeping it alive for 24 hours? It would be dormant most of the time, mostly active between 6pm and midnight, but there are occasional games at odd hours. Is this something better handled by just making my script a systemd service instead of using a scheduler like Prefect?

Hello, there seems to be a bug with Automations and flow run state. I have an alarm to notify me if a flows stays in the running state for more than 4 hours. However, I now get notified for all flows that have completed, 4 hours hours later. Is there any bug related to that ?

Hi, i’m new to Prefect and it looks promising. Basically i’m trying to build a standard toolkit for deploying pipelines in production, with Kubernetes. I read the documentation but some points weren’t clear to me. 1. Does Prefect support custom reusable components and a library of it for users to choose from? (E.g. Supports some form of definition and allows checking of interfaces) 2. Does Prefect support building pipelines using UI? Thus far it seems very pythonic. 3. Say i would want to deploy a pipeline with Prefect and using Kubernetes as the infra, can i ask Prefect to do it on a single k8s node? Reason for doing that is to create a low latency pipeline. 4. Does prefect deployment allow HPA?

Hi folks, I'm trying to figure out the simplest and least expensive way to get my first Prefect flow into production. Am I correct in thinking I could just spin up a Digital Ocean $5/mo droplet, install

prefect

on it, copy my workflow repo onto it,

prefect login

to prefect cloud, and then have systemd run

prefect agent start

? Or do I need to add a 'process block'?

Hi, I’m new to Prefect. Is there a preferred way to run R scripts in a flow?

Hi, I'm trying to load csv files from gcs to bigquery. I used

bigquery_load_file

method, but it can't find the dataset in bq. I am sure I have created the dataset.

HI everyone , I have a bit of confusion on what infrastructure subflows are run on? The documentation is not very clear regarding this. E.g. I have deployed agents on self hosted kubernetes on multiple Machines running in different regions of a cloud provider (E.g 3 machines running on EU, ASIA and US) all the agents are querying the same queue. If the parent flow is run on EU, does the subflow run on 1. The same pod 2. Goes through the queue again and can be picked by any agent running in any region To rephrase, does subflow run create a fork/spawn in the same infrastructure or go through the queue by inheriting the deployment infra blocks?

Hello! I’m experiencing problems with subflows and

timeout_seconds.

I’ve put a timeout on both the parent and sub-flows, and the timout on the parents works nicely, leading to a ‘Timeout’ state after the specified time. The sub-flows just go on forever though. Anyone know what might be going wrong? Prefect version

2.7.8

I'm trying to implement a new flow to run with container. But I'm getting this error when deploying. How do I resolve this?

Traceback (most recent call last):
  ...
  Remote Storage block must have 'get-directory' capabilities. (type=value_error)

Hi! If I want to have user specific access to Secrets in prefect cloud I need to use enterprise subscription, correct?

Is there an integration between Meltano <> Prefect as there is with Airflow? https://docs.meltano.com/guide/orchestration

Hello, Just exploring Prefect 2.x. Currently looking at on_complete and on_failure hooks. Looking at logging within these blocks of code. If I call get_run_logger I get "prefect.exceptions.MissingContextError: There is no active flow or task run context." and if so a simple print it does not show up in the log. Is there a way for logs in these hooks to be logged? Thanks

Are there any data catalog/lineage tools that integrate with Prefect?

Hey ya’ll! I’m currently looking into using Prefect as a solution for our DE & ML team- I’ve got a quick question about project structure. Is there any documentation on preferences or recommendations on how Prefect users should organize their projects (projects, flows, tasks, utility files, etc…)? Thanks!

is there a good way to add a standard set of parameters to all flows in prefect 1? we're trying to do something like this

class DHFlow(Flow):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.add_task(Parameter(name="some_param", default=True))

which works locally but breaks in the cloud with

Failed to retrieve task state with error: ClientError([{'path': ['get_or_create_task_run_info'], 'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'locations': [{'line': 2, 'column': 101}], 'path': None}}}])
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine/cloud/task_runner.py", line 154, in initialize_run
    task_run_info = self.client.get_task_run_info(
  File "/usr/local/lib/python3.10/site-packages/prefect/client/client.py", line 1803, in get_task_run_info
    result = self.graphql(mutation)  # type: Any
  File "/usr/local/lib/python3.10/site-packages/prefect/client/client.py", line 570, in graphql
    raise ClientError(result["errors"])
prefect.exceptions.ClientError: [{'path': ['get_or_create_task_run_info'], 'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'locations': [{'line': 2, 'column': 101}], 'path': None}}}]

Hey everyone! Does anyone know a way to use Python to get a list of all JSON blocks from my Prefect Cloud workspace? I want a list of all the names, assuming I don't know any.

Hi all, We are exploring prefect as a data engineering platform tool. The company doesn’t want to use EC2 instance. Wanted to ask how can we deploy prefect on AWS.. And can we configure prefect ui authenticated via SSO .. we use OKTA.. Are there any tutorials or documentation for above ?

Hi everyone, Last night we upgraded prefect version to 2.8.4 We keep getting these errors in prefect

Flow run infrastructure exited with non-zero status code -1.

But when I check the logs of kubernetes pod, tasks keep running. Sometime even the subsequent tasks are triggered but at time it fails and kills the pod. Anyone knows what can it be? Thanks.

I'm using Automation to trigger an action if a flow-run with a certain tag is executed. I set Trigger type = Flow run state, Flows = All Flows, Flow Run Tags = local and Flow Run = Enters Any state. The name of my tag is "local". The automation works if I create a deployment for my flow and start flow runs from there. However, if I start a flow run locally then this automation does not trigger. Is that expected? The main body of my flow is this:

if __name__ == "__main__":
    with tags("local"):
        my_flow()

In the UI I can see all flow runs, whether run through the deployment or locally. All have the required "local" tag assigned to it. I tried removing the tag from the automation and select my flow name instead. Then the automation works both when run as a deployment and when run locally. Is this a known issue or limitation in using tags for an automation?