2) How to handle multithreading inside custom modules? Task runs from Prefect, but code defined in custom module. If there is a multithreading block like below, Prefect will not wait until all futures/threads are completed, but succeed task immediately and exit.

# with concurrent.futures.thread.ThreadPoolExecutor(max_workers=10,
#                                                   thread_name_prefix="flow_") as executor:
#     futures = [executor.submit(self._run, df) for df in self.dfs]
#
#     for future in as_completed(futures):
#         try:
#             future.result()
#         except Exception as exc:
#             logger.exception(exc)

Normally with Prefect, you shouldn't have to use multithreading since when you use mapping, Prefect does that for you automatically

Yeap, for DaskCluster its clear. It controls multithreading itself and that all things work fine. I was asking in the context of our library has built in multithreading features that allows you to submit tasks from any command line, but when you take that library as is and run in Prefect - that won't work. So basically we need to wrap each library function into Prefect Task and then pass them through Prefect mapping feature

Coming from other thread, if you pass the

logger

as an input into your multiprocessing function, it will work, but really just use LocalDask and then you get observability into your tasks as well

Yeap, for DaskCluster its clear. It controls multithreading itself and that all things work fine.

Actually, using

LocalDaskExecutor

doesn't use Dask, it uses multithreading. Check out https://discourse.prefect.io/t/what-is-the-difference-between-a-daskexecutor-and-a-localdaskexecutor/374