Hello there, a question about Prefect 2.0 and flows using Dask runner. I posted it over at data-tricks-and-tips but this places seems to be more active. Hoping somebody can help with this! https://prefect-community.slack.com/archives/C03CY667GGM/p1660137898917299 It relates to getting the Prefect's task's Dask worker client from inside the task. Using

worker_client

context manager works as expected in Prefect 1.0 (see https://discourse.prefect.io/t/how-to-use-dask-without-mapping-in-prefect-1-0-using-das[…]-client-to-call-client-submit-inside-a-prefect-task/470/13) but this behavior doesn't seem to migrate over to 2.0

Hey folks, Is anyone else having issues with Prefect Cloud UI. I have black panels on the display that weren’t there before, and contents in certain places not showing up. Tried different browsers and it doesn’t appear to help.

Hey everyone, I have a Prefect (1) flow that uses a GPU (Nvidia T4). I get this really odd behavior - I register the flow using python 3.9, and I run it on a python 3.8 container (tensorflow/tensorflow:latest-gpu) When running on a machine without GPU - everything works. When running on a machine with GPU - I get this error -

Task 'upload_data_to_bq_task': Exception encountered during task execution!
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/prefect/engine/task_runner.py", line 880, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/usr/local/lib/python3.8/dist-packages/prefect/utilities/executors.py", line 468, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/Users/dekelr/PycharmProjects/similarity-filter-layer/prefect_tasks/upload_data_to_bq.py", line 24, in upload_data_to_bq_task
SystemError: unknown opcode

The code fails when running this specific row - (batch size is either an int or None)

if batch_size is None:

Everything works fine when I change this row to “if not batch_size:” After some troubleshooting - I found this thread - https://github.com/PrefectHQ/prefect/issues/3635 Running the same flow with the original row of -

if batch_size is None:

Still doesn’t work when registering with python 3.8 (same python version as in the container) Can you please explain this really odd behavior? Thanks

Hello, we tried to install redshift connector into docker and even though it was successful, every time we run the script using the redshift connection, prefect dev returns module not found error.

I'm having some trouble running deployments that override a flow's default parameters, has anyone bumped into the same issue? • I have a flow that takes a parameter, the parameter has a default value (let's call it False). • Before running deployment apply, I modify the deployment.yaml file so that the parameter has the value True instead. • When viewing the deployment in the GUI, it correctly states that the parameters default value is True (based on what was in deployment.yaml) • However, when I run the deployment and log the parameter, the flow run reports the parameter value as being False. I have tried a few different data types in order to see if the behavior changes, such as bool and str, but it makes no difference. I'm using Prefect cloud and I'm on 2.0.4 with a fresh venv.

I am getting the above error when trying to run a flow in Prefect 2.0. I'm in namespace dev-prefect in k8s. The flow logs in the Prefect UI are completely blank. Do I need to make a custom service account in k8s?

Hello, Can we configure the Orion Database for Prefect Cloud ? Or is it already managed by Prefect cloud ? I also want to ask about the sensitivity of the data stored there, as there will be task results stored in the database. So how does it work ?

I'm using Prefect 2 to build a data pipeline and have a couple of general design questions: • I assume we should be thinking about holding data in S3 and passing it between flows by reference – i.e. rather than as parameters • If so, should I use S3 blocks to store intermediate results, or interact with S3 directly? What are the trade-offs? • I'm planning on using Task.map to parallelise work (using dask); there's no equivalent for Flows, so I guess we should be thinking of parallelisation only happening within flows, is that right?

hello! does anyone have a graphql query to cancel all running instances of a specific flow? i’m on prefect 1.0 if that makes a difference

Hi. Any plans to update this terraform module for 2.0? https://github.com/aws-ia/terraform-prefect-agent-ec2

👋 are there any plans for prefect 2.0 to hook into common apm tooling (e.g. datadog, open telemetry)? It would be great if i could get all my tasks instrumented and reporting into datadog apm so i could, for example, see a distributed trace between a prefect flow/task and the various services it interacts with. I know datadog has a ddtrace library that is maybe a starting point?

Greetings, I am trying to install the old prefect-gcp for prefect 1.0 to execute BigQueryTasks

prefect-gcp

seems to only refer to Prefect 2.0

Do tasks need to be in the same file as the flow which uses them 🤔 ? When I had a larger task in a separate module, I got PicklingErrors (stack trace in thread) – if I move the task definition into the my main module it works as expected.

Hello. If I want to run flows in a local subprocess. Do I have to specify the environment in which prefect agents will execute the tasks? Let's supposed my flow requires some dependencies that I have already installed in a python environment. When the agents will pick up the flow, where do I specify the environment they need to in other to run the flow?

Hey, I just upgraded from prefect 2.0.0 to 2.0.4 and re-created my deployments. For one of my old deployments, I manually added a schedule-Definition and renamed the file to mydeployment_old.yaml. Then I ran prefect deployment build myflow.py:myflow --name mydeployment -t mytag -sb gcs/gcs-prefect-stprage -o mydeployment.yaml And like magic, the schedule definition block was automatically added to mydeployment.yaml. This is by the way awesome - but ... I don't know why this happened 😅 So, my question is I guess: How does prefect build know, which deployment-yaml settings to apply? Did it simply look in the mydeployment_old.yaml file which was in the same folder and figure "oh well, there is a schedule block in there, so I guess he'll need it in the new deployment as well"? (To get my intention right: This feature is awesome, I'd just like to understand how exactly it works, so that I can use it as intended)

We actually got our Prefect 2 flows working in our production environment yesterday! 🎉 However, today's big daily scheduled orchestrator job has a problem I haven't seen during my 2.5 weeks of development and testing: Two of the created flows do not get picked up by the agent, they're stuck in the work queue. I just tried to start another deployment and now there's 3 in the queue. All the previous flows were picked up correctly and they all have the same tags. Has anyone run into a similar problem? I'm not sure what other information should include.. The agent is still running. There are no concurrency limits set for this work queue.

Hi all, I have a problem with prefect-snowflake and/or concurrentTaskRunner and/or asyncio: I'm trying to run ~10 snowflake queries simultaneously (example code in thread) And only 4 or 5 or 6 are picked, the rest queries run when some previous query finishes. It looks that the threads do not "go to sleep" as intended when they wait for snowflake results and because of that other tasks are not picked.

Hi everyone! I have the following issue, that I cannot seem to set the secret api key correctly inside the agent: I have verified with this: kubectl get secret api-key --namespace prefect2 -o jsonpath='{.data.prefect_api_key}' | base64 --decode This is done in Azure following Cristopher Boyd:

Hey. I've an issue with using the prefect-airbyte collection (v. 0.1.0) with prefect 2.0.4 (not related to version 0.4, it also occurs with any prefect 2 version). If this is the wrong channel to ask about collections, let me know. I've a very simple flow, which triggers one or many airbyte connections. The flow is almost 1 to 1 from the doc-samples. If I run the flow with triggering one airbyte connection id and wait, until the flow is done - and then wait for 10 (something) more seconds - before I start another flow run, everything is fine. However, if I run the flow twice in parallel (no matter whether it's with the same airbyte connection - id or a different one), I get a TaskCanceled Exception (see thread) - but the airbyte connection still gets triggered and finishes successfully. So the flow triggering airbyte works, but the flow gets canceled seconds after the airbyte connection is triggered. As the airbyte connections get both triggered (and both finish successfully), I'm starting to ask the question here - but I can imagine, that this might also be a problem on the airbyte side... Any hints, how to circumvent this? Edit: The same result whether I run the flow locally or use a dedicated agent to run it.

Is it recommended to use a Postgres backend when using Dask? I am getting the following errors with the default SQLite implementation.

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) unable to open database file

Hi all. Can someone point me to documentation on how to have lazarus restart dead flows from the beginning? It appears the default is to restart from the failed task.

It looks like I can no longer have keyword function params defaulted to

None

defined on a mapped task? I now get this error (

TypeError: object of type 'NoneType' has no len()

) when trying to map across the subsequent function signature.

@task
def get_items_list(
    client, entity, page,
    response_attr=None, path=None, params=None
):

I've seen other questions about this, but not a solution yet. When using

Secret.load("slackbot-oauth").get()

, I get

AttributeError: 'coroutine' object has no attribute 'get'

whether I run this in an ipynb or with a python file through the terminal. Is it expected to run this command with await and async? Or is this not supposed to happen?

Is there a prefect 2 equivalent to template task names? I've searched high and low on the prefect 2 docs/slack and can't find anything. old docs: https://docs-v1.prefect.io/core/concepts/templating.html#templating-names

hi all, I've been in transition to prefect 2 (2.0.4 to be precise) from prefect < 2 (1,2 to be precise). Thanks to your help I sovled most of the problems, only one remains: in prefect < 2 I used parallelism via

multiprocessing.Process

as

processes = []
    for chunk in url_chunks:
        kwargs["mp3_urls"] = chunk
        processes.append(mp.Process(foo=foo, kwargs=kwargs))
    for p in processes:
        p.start()
    for p in processes:
        p.join()

and it worked. The main reason for using

multiprocessing.Process

is to have a large model in memory shared between processes. In fact it works when I run this exact flow with prefect via a python script calling the flow as a fucntion, it works as expected. However, when run it as

prefect deployment run ...

or via a scheduled run, the processes in

p.start()

don't run. The flow result is marked as success to my surprise. Any ideas?

my colleague can't login to prefect 1.0 due to a forgotten password, but he's unable to reset his password as it says to contact his system admin (me) - I have received no email regarding his password reset request @Dave Thomas

Hi, I am trying to run a

KubernetesAgent

locally and to bind it to minikube for testing purposes. For now, I see the Agent in the Server UI, but the tasks are not executed. How do I configure this? Thank you for you patience, I am new to Prefect.

😄  minikube v1.26.1 on Ubuntu 20.04
✨  Using the docker driver based on existing profile
👍  Starting control plane node minikube in cluster minikube
🚜  Pulling base image ...
🏃  Updating the running docker "minikube" container ...
🐳  Preparing Kubernetes v1.24.3 on Docker 20.10.17 ...
    ▪ Generating certificates and keys ...
    ▪ Booting up control plane ...
    ▪ Configuring RBAC rules ...
🔎  Verifying Kubernetes components...
    ▪ Using image <http://gcr.io/k8s-minikube/storage-provisioner:v5|gcr.io/k8s-minikube/storage-provisioner:v5>
🌟  Enabled addons: storage-provisioner, default-storageclass
🏄  Done! kubectl is now configured to use "minikube" cluster and "default" namespace by default

Getting an SQLAlchemy error when trying to parallelize about 3000 task runs on a Dask cluster. Confusing to me as this is something that "just worked" on Prefect 1.0, but isn't working for me on 2.0...

sqlalchemy.exc.TimeoutError: QueuePool limit of size 5 overflow 10 reached, connection timed out, timeout 30.00 (Background on this error at: <https://sqlalche.me/e/14/3o7r>)

Hi I am just setting up prefect for my Data engineering team ran into one issue when trying to setup ecs fargate.

prefect agent ecs start --cluster arn:aws:ecs:us-west-2:00000000:cluster/prefect-dev --label aws-ecs

Usage: prefect agent [OPTIONS] COMMAND [ARGS]...

Try 'prefect agent --help' for help.

╭─ Error ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮

│ No such command 'ecs'.                                                                                   │

╰───────────────────────────────────────────

Hey - are you guys recommending prefect 2.0 for production yet? Is there an ECSRun type setup that I can explore in order to replicate what we currently have in 1.0 ? Thanks