Hello, I am having trouble importing prefect flow and task in a databricks notebook.

from prefect import flow, task, Parameter

First error was "ImportError: cannot import name 'deprecated' from 'typing_extensions'" when I was using python 3.10. I changed the version to 3.8 and now getting "ImportError: cannot import name 'TypeGuard' from 'typing_extensions' (/databricks/python/lib/python3.8/site-packages/typing_extensions.py)" Here's the full error log. How can I resolve this?

ImportError                               Traceback (most recent call last)
<command-1880743847103486> in <module>
      1 from typing import Any, Awaitable, Dict, List, Optional, Union
----> 2 from prefect import flow, task
      3 from prefect import Parameter
      4 from typing import Any
      5 from datetime import timedelta

/databricks/python_shell/dbruntime/PythonPackageImportsInstrumentation/__init__.py in import_patch(name, globals, locals, fromlist, level)
    165             # Import the desired module. If you're seeing this while debugging a failed import,
    166             # look at preceding stack frames for relevant error information.
--> 167             original_result = python_builtin_import(name, globals, locals, fromlist, level)
    168 
    169             is_root_import = thread_local._nest_level == 1

/databricks/python/lib/python3.8/site-packages/prefect/__init__.py in <module>
     35 
     36 # Import user-facing API
---> 37 from prefect.states import State
     38 from prefect.logging import get_run_logger
     39 from prefect.flows import flow, Flow

/databricks/python_shell/dbruntime/PythonPackageImportsInstrumentation/__init__.py in import_patch(name, globals, locals, fromlist, level)
    165             # Import the desired module. If you're seeing this while debugging a failed import,
    166             # look at preceding stack frames for relevant error information.
--> 167             original_result = python_builtin_import(name, globals, locals, fromlist, level)
    168 
    169             is_root_import = thread_local._nest_level == 1

/databricks/python/lib/python3.8/site-packages/prefect/states.py in <module>
     10 import httpx
     11 import pendulum
---> 12 from typing_extensions import TypeGuard
     13 
     14 from prefect.client.schemas import State as State

ImportError: cannot import name 'TypeGuard' from 'typing_extensions' (/databricks/python/lib/python3.8/site-packages/typing_extensions.py)

Hey guys, I’m running a flow with a DaskTaskRunner that spawns a task that takes around 20seconds and apparently it’s running sequentially?! My machine has 16 threads and I see in the UI each task only being run after another one finishes. Any inputs?

23:04:54.057 | INFO    | prefect.engine - Created flow run 'optimal-snail' for flow 'test1'
23:04:54.057 | INFO    | prefect.task_runner.dask - Creating a new Dask cluster with `distributed.deploy.local.LocalCluster`
23:04:55.980 | INFO    | prefect.task_runner.dask - The Dask dashboard is available at <http://127.0.0.1:8787/status>
23:05:00.371 | INFO    | Flow run 'optimal-snail' - Created task run 'Execute values of the query-8165e3c8-0' for task 'Execute values of the query'
23:05:00.372 | INFO    | Flow run 'optimal-snail' - Executing 'Execute values of the query-8165e3c8-0' immediately...
23:05:24.612 | INFO    | Task run 'Execute values of the query-8165e3c8-0' - Finished in state Completed()
23:05:26.370 | INFO    | Flow run 'optimal-snail' - Created task run 'Execute values of the query-8165e3c8-1' for task 'Execute values of the query'
23:05:26.370 | INFO    | Flow run 'optimal-snail' - Executing 'Execute values of the query-8165e3c8-1' immediately...
23:05:42.533 | INFO    | Task run 'Execute values of the query-8165e3c8-1' - Finished in state Completed()
23:05:44.295 | INFO    | Flow run 'optimal-snail' - Created task run 'Execute values of the query-8165e3c8-2' for task 'Execute values of the query'
23:05:44.296 | INFO    | Flow run 'optimal-snail' - Executing 'Execute values of the query-8165e3c8-2' immediately...
23:06:09.538 | INFO    | Task run 'Execute values of the query-8165e3c8-2' - Finished in state Completed()

As you see in these logs it’s not really spawning more tasks. I was under the impression that the call to a function with the

@task

decorator would be non-blocking if made within a flow

@flow

which uses DaskTaskRunner

In Prefect we can define tasks as functions with

@task

decorator, or as classes by subclassing

Task

. In general when programming in python and presented with these options, I typically reach for the subclassing option. I find I can think about problems at a higher level with OOP, easier to build up complex components from simpler ones, cleaner abstraction, etc. However I do find it’s a bit clunky with at times when building flows. Is it recommended to avoid the subclassing approach and use a functional approach instead when possible?

What is the best approach to calling an async function within a Prefect task? I have to call the Azure Eventhub client which is async all the way down, from my straight synchronous Prefect task. I've tried calling the async functions like so;

asyncio.run(client.send_batch(event_data_batch))

but I get errors like

RuntimeError: Event loop is closed

Any advice?

I'm new to Prefect, and have been experimenting with running flows on my local machine with a local server as well. So far things have been very intuitive! However I have another nodejs app running locally, and want to try to have that app call the local Prefect server to start a parameterized flow. I can start flows through the UI at localhost:8080 and I've been going through the documentation but things just haven't clicked and I'm not sure how to do this. (Additional disclaimer - I have virtually no experience with GraphQL which probably isn't helping). Thanks for the help and this amazing product!

Hey all, has anybody successfully started a dask cluster + prefect on top of an existing spark cluster (databricks)? I have a databricks cluster but want to use distributed prefect to schedule

Hi all, after successfully utilizing an ecs:push workpool with my own provisioned AWS infrastructure to deploy my flow to and run it on the ecs cluster, I was curious about the new feature to let prefect provision the infrastructure 😃 Unfortunately, it doesn't work as expected. I created the ecs: push work pool as suggested (https://github.com/PrefectHQ/prefect/pull/11267) and after that I deployed my flow to it as usually, but without job-variables (since I don't have to ingest the AWS infrastructure and role arns now). I want to push the flow image to my ecr repo, so I give this with the name of the image to my Deployment like this:

flow.deploy(
    ...
    image=DeploymentImage(
        name=os.getenv("ECR_REPO_URL", ""),
        tag=os.getenv("IMAGE_TAG"),
        dockerfile=cfd / "Dockerfile",
    ),
    ...
)

But then I get the error: Flow run could not be submitted to infrastructure: An error occurred (ClientException) when calling the RegisterTaskDefinition operation: Fargate requires task definition to have execution role ARN to support ECR images. Why do I still have to provision the deployment with an execution role, shouldn't it (or the work pool) create one? Or is it because it is a my own ECR Repo? Where do you usually push / save the flow image to run it on the ecs cluster (especially in the frame of the ecs:push work pool with infra provisioning)? Thank you and best regards!

@Marvin How do I change the state of a flow to Failed via the API in Prefect 2?

@Marvin how can I concatenate a prefect string block with an additional string in a prefect.yaml file?

Does anyone have a complete working example on how to use PrefectResult? Its simple: Scrape a website for a link, if the link is different than the prefect result, download it and save the new link as the result. Otherwise exit with "success: no work to be done". I don't want this to be a local result because we plan to move these flows off of a windows agent and into docker agent. Thanks in advance!