import time

from prefect import flow, task
from prefect_dask import DaskTaskRunner

@task
def shout(number):
    time.sleep(5)
    print(f"#{number}")

@flow(task_runner=DaskTaskRunner)
def count_to(highest_number):
    for number in range(highest_number):
        shout.submit(number)

if __name__ == "__main__":
    count_to(10)

import time

from prefect import flow, task
from prefect_dask import DaskTaskRunner

@task
def shout(number):
    print(f"before sleep")
    time.sleep(5)
    print(f"#{number}")

@flow(task_runner=DaskTaskRunner(cluster_kwargs={"n_workers": 4, "threads_per_worker": 2}), log_prints=True)
def count_to(highest_number):
    cats = ["fluffy", "loaf", "fat", "squishy"]
    for cat in cats:
        shout.submit(cat)

if __name__ == "__main__":
    count_to(10)