Hi, I am just starting to use Prefect for workflow management ( migrating from Celery). I have few basic questions, 1) Is it possible to run a task asynchronously just like some_celery_task.delay()? 2) Is it possible to deploy Prefect flows without using Prefect cloud server? 3) For periodic tasks using Prefect schedules, it appears that flow.run(schedule=periodic_schedule) is a blocking call. Is there something like celery periodic task which gets triggered automatically in the background? Thanks

Hi @Gopal — great questions. We don’t support Celery (yet!) but we have first-class support for Dask, which is a similar system more optimized for data science use cases. By running Prefect with a Dask executor, you get speed and scale benefits as well as asynchronous execution.

flow.run()

is indeed a blocking call (it’s really more of a convenience method for some of the low-level machinery), but when paired with Dask the actual execution will be asynchronous and remote.

@Jeremiah That is of great help. So the dask executor is the key here for the asynchronous execution. For periodic asynchronous execution do you suggest to use Cron as of now?

It depends how asynchronous you need to be. If you can run a single, synchronous process that “launches” asynchronous flow runs, then you could attach a

prefect.schedules.IntervalSchedule

or

prefect.schedules.CronSchedule

to your flow and simply call

flow.run()

with a

DaskExecutor

. The

flow.run()

call will block, but the flows will run in the dask cluster. If you require fully asynchronous (each flow launched discretely, no blocking process) then Cron might do it.

ah. That is a good idea. For my requirement a single master flow which is periodic with the async task flows should be good enough. I will try that. Thanks for your quick response 🙂

Let us know how it goes!

@Jeremiah flow.run(executor=DaskExecutor(address='tcp://192.168.34.50:8786')) is still blocking. Is there something like "dask.delayed" in Prefect? My use case is running a task upon an API request without timeout. The API will just create a flow and run using Dask executor. However, the flow.run() waits till the task finishes. Am I using Prefect wrong?

flow.run()

is always a blocking call because the

FlowRunner

is doing state management. I’m tempted to say you could simply submit

flow.run()

itself to your dask cluster (create a dask client and

distributed.fire_and_forget(client.submit(flow.run))

) but cc @Chris White in case that creates an issue with dask worker clients…

I see. Fire and forget works atleast on the local dask workers. I will check with the deployment. Thanks for your suggestions. @Chris White please let me know if you see potential issue with fire and forget

Our Dask executor is generally smart about detecting whether it’s on a worker or not, so I think this should work OK

Thanks for the clarification !