how do i set/access environment variables with a dask executor? i had been setting them with my runconfig, but when i switched to a dask executor they arent showing up anymore

Hi, Could you please help with configuring DockerFlowRunner to access flows from GCP storage? I’m following Perfect 2.0 tutorial. I configured GCP storage and successfully used it with “universal runner” for non-dockerised flows. But when I try to run flow deployed with a DockerRunner it fails with the following error cause it couldn’t access the GCP(exception is in the thread). I’m using default user authentification for GCP. Setting

GOOGLE_APPLICATION_CREDENTIALS

env var to point to a default location (

/Users/test/.config/gcloud/application_default_credentials.json

) before running an agent didn’t help.

can we run workflow inside workflow, like

def workflow1():
   @task
   def abc(): somelogichere

   with Flow('one') as f:
      abc()
   
   f.run()

@task
def xyz(): workflow1()

with Flow('main flow') as f:
   xyz()

f.run()

Hey all, I've been playing with running Prefect 2 in a docker container on Google Compute Engine. I created a basic "hello world" example that uses Github Actions to build the container and deploy to Compute Engine. Let me know if you have any suggestions and if this is helpful: https://github.com/BennyJW/prefect2-docker-gce

When we delete flows using the graphql mutation; how do we ensure that all versions of a flow are deleted and not just the most recent / active version?

Hi, I'm looking to integrate with a database and we are exploring putting some data into a CockroachDB database, but I did not see it in the prefect integrations. Do you recommend instead using a PostgreSQL database?

Hi, posting incase anyone else is having issues in their pytest CICD pipelines after updating to the awesome beta4 release of 2.0: It seems that the

prefect_test_harness

was moved into a new module (

prefect.testing

), that is not available in 2.0b4 - even though it is available in the orion branch in the repo. I created an issue at GH also: https://github.com/PrefectHQ/prefect/issues/5787

Hello! Hopefully a simple question: I am trying to create a flow with two schedules (let's say A and B). Is there any way to specify an additional flow label for each schedule (e.g., 'a' for A schedule and 'b' for schedule B)? What I ultimately would like to achieve is to enable the same flow to be run on different agents at different times - maybe there is a better way? P.s. I'm using Prefect 1.0

seeing errors like

Error during execution of task: KeyError(<Thread(Dask-Default-Threads-12-578, started daemon 140412823688960)>)

when retrying tasks run on dask. is there special configuration i need to do for a retry with dask?

Hi all, we've just upgraded the python version to 3.10 and re-registered all the flows but we are getting this error, could someone assist?:

Exception raised while calling state handlers: SystemError('unknown opcode')
Traceback (most recent call last):
  File "/mnt/data/prefect/venv/lib/python3.8/site-packages/prefect/engine/cloud/flow_runner.py", line 119, in call_runner_target_handlers
    new_state = super().call_runner_target_handlers(
  File "/mnt/data/prefect/venv/lib/python3.8/site-packages/prefect/engine/flow_runner.py", line 116, in call_runner_target_handlers
    new_state = handler(self.flow, old_state, new_state) or new_state
  File "/mnt/data/prefect/venv3.10/lib/python3.10/site-packages/prefect/utilities/notifications/notifications.py", line 65, in state_handler
    def state_handler(
SystemError: unknown opcode

Hi all, I'm seeing a new error crop up in a flow I've been running stably for a few months. The flow is a top-level pipeline that runs a series of

create_flow_run

(and

wait_for_flow_run

) tasks. One of these steps is followed up with a

get_task_run_result

, which has always worked without issue until today. Now it's throwing this error:

ValueError: The task result cannot be loaded if it is not finished

. I'm not seeing how this could be, since I can see in the logs that the upstream task did in fact finish successfully. I tried explicitly setting the result of

wait_for_flow_run

as an upstream dependency of

get_task_run_result

(which i think shouldn't be needed), and I also tried setting the

poll_time

to

30

, but still no luck. Does anyone have any ideas?

I am in the process of building CI/CD pipeline using prefect 2.0 for cloud. I am running into an issue where i get the error:

You have not configured default storage on the server
or set a storage to use for this deployment but this deployment is using a 
Kubernetes flow runner which requires remote storage.

I have already defined the s3 bucket as the storage in prior steps and make sure to even reset it as the default before hand. I have no problem creating the deployment locally only an issue when running it on github actions.

Got a questions in regards to handling DST correctly. If i specify the anchor_date in winther time, it will be offset by 1 hour. is there a way to handle this correctly?

import asyncio
import pendulum
from datetime import timedelta
from prefect.orion.schemas.schedules import IntervalSchedule

winter_schedule = IntervalSchedule(
   interval=timedelta(hours=24),
   anchor_date=pendulum.datetime(2022, 1, 1, 0, 30, 0, tz="Europe/Copenhagen")
)

summer_schedule = IntervalSchedule(
   interval=timedelta(hours=24),
   anchor_date=pendulum.datetime(2022, 4, 1, 0, 30, 0, tz="Europe/Copenhagen")
)


print(asyncio.run(winter_schedule.get_dates(1))[0])
print(asyncio.run(summer_schedule.get_dates(1))[0])

>>> "2022-05-16T01:30:00+02:00"
>>> "2022-05-16T00:30:00+02:00"

Hello am new to prefect and just reading through the docs. On the Flows page this is super confusing to me:

Return a future

If a flow returns one or more futures, the final state is determined based on the underlying states.

```from prefect import task, flow

@task

def always_fails_task():

raise ValueError("I am bad task")

@task

def always_succeeds_task():

return "foo"

@flow

def always_succeeds_flow():

x = always_fails_task()

y = always_succeeds_task()

return y```

What does it mean by futures here? My understanding was that futures are to do with threading and async in Python, of which I see neither here? What am I missing? 🤔

I am using some deep learning model inside a task in the flow. The flow is configured with docker agent. It is downloading the model from web everytime I run the flow. How to load the model from local system where the agent is running?

Hey all, I’ve been looking into Prefect as a potential tool for a streaming data pipeline, and I feel like I might be misunderstanding something about the pricing model, so I’m wondering if someone can help me sort it out. After reading [this blog post](https://www.prefect.io/blog/you-no-longer-need-two-separate-systems-for-batch-processing-and-streaming/), I was pretty excited about trying Prefect for this use case. When I looked at the pricing page, I saw that the unit being used to measure usage was task runs. So I guess I have two questions about that: 1. Is it fair to say that an implication of paying for task runs is that one would want to minimize the number of tasks in their flow, so as to get the most “bang for their buck?” This seems counter-intuitive to me; I’m on Page 3 of the Prefect Docs, “Thinking Prefectly”, and one thing I’m pretty sure I know by now is that smaller and more discrete tasks are better. Plus, that blog post (rightfully) pointed out that a Flow is better orchestrated by breaking up the logic into multiple tasks as needed. 2. Is it also fair to say that given that a streaming data pipeline pulls events from a stream at some chosen interval, and presumably each pull constitutes at least one “task run”, the cost of the Prefect job is proportional to the rate at which data is processed? For example, is running a Flow which is efficiently able to pull and process batches of messages every 2 seconds going to be twice as expensive as a Flow which is pulling and processing new batches every 4 seconds? (by the way, when I plug the number of monthly task runs equivalent to continuously pulling batches of messages every 2 seconds into the Pricing calcuator, it comes out to… Contact Us 😄)

Do you have any date about when the development of Orion will happen on the open repo on Github?

Hi - is there a way to change the logging format when using Prefect Cloud? Setting

PREFECT__LOGGING__FORMAT

in the flow config env doesn’t seem to work.

hi community! i'm trying to set up bitbucket storage (using bitbucket cloud) for my flows (working off this doc https://docs.prefect.io/api/latest/storage.html#bitbucket), but i'm getting a 404 when trying to execute the flow

404 Client Error: Not Found for url

. I've tried various combinations of repo/project/auth credentials for Bitbucket storage, but all with the same issue - has anyone successfully set up Bitbucket storage? (using Prefect 1.1.0)

Hi there. Could someone help me understand why some flow runs get stuck? This shows on Prefect Cloud as if it has been running since May 4 but the process is no longer running. Thanks https://cloud.prefect.io/flow-run/a1f0f6df-793d-4fa8-af53-a7f586eadabc

Hi, I would like to use a custom image from my private image registry to run a flow on my kubernetes flow runner (orion). How can I provide credentials for this?

hey, might be a dumb question but where could i find the changelogs for Prefect Orion ?

Hi when I am raising a FAIL signal the logger logs with the INFO level. it is possible to get it to log with the ERROR level instead?

raise signals.FAIL(message=f"Task failed") # This should log with ERROR level, not INFO

Hy I am trying to return list of object from aws s3 bucket. I am getting this error when using it in prefect flow “TypeError: can’t pickle _thread.lock “objects using s3_client.list_objects(bucket,prefix) Otherwise its returning correct output without prefect on jupyter notebook

Hi everyone! I encountered an error when attempting to restart a flow from the failed task and was wondering if anyone has some best practices on how to handle this? Context: • I have a "Flow of Flows" that triggers a few connectors from Fivetran in parallel then triggers a dbt job run for the data modelling for the source once the data has finished syncing • There is one dbt job that is then run after all the prior dbt jobs have succeeded (this is the one that failed in this instance) • After fixing the error, when I press restart I get an error saying ◦ This then returns a list of all the flow runs on this flow id Question: • What would I need to specify in the task parameter so that it only re-runs the latest flow when I press restart on the UI? Thank you so much in advance for your time and help! 😄

Hi! I have a project that started with flow.py, now I need to create another flow, is it ok that I just create the new flow (db-flow.py) in this same folder?

Hello, I'm trying to trigger a flow using an AWS S3 PUT event - through Lambda function. Whenever there is an event, flow get scheduled but is stuck there. I have gone through other messages in the community - but there is no problem with labels as well. Can anyone help me with this? I used prefect cloud. Thank you.

im getting dask timeouts with my temporary clusters. im running dask in gke with prefect cloud, and im seeing a flow hang on

Creating a new Dask cluster with `__main__.get_executor.<locals>.<lambda>`...`

in my gke logs, im seeing

raise TimeoutError( asyncio.exceptions.TimeoutError: Nanny failed to start in 60 seconds

When calling

future.get_state()

, a 403/timeout response will cause flows to fail - is there a way to see api limits for prefect orion cloud? We get forbidden (403) occasionally - it appears to be due to throttling

Hey, I’m trying to configure Prefect’s logger and add a Logz.io handler. Can’t find any good example - can anyone help me with this? I configured a logger using this documentation https://docs.logz.io/shipping/log-sources/python.html but I can’t make Prefect’s “prefect.context.get(logger_name)” use it. Thanks.