Prefect Examples with RunConfigs and Executors

ecs_with_dask_cloudprovider.py

from prefect import Flow
from prefect.storage import S3
from prefect.run_configs import ECSRun
from prefect.executors import DaskExecutor

with Flow("example") as flow:
    ...

flow.storage = S3("my-flows")
flow.run_config = ECSRun()  # Run job on ECS instead of locally
flow.executor = DaskExecutor(
    cluster_class="dask_cloudprovider.aws.FargateCluster",
    cluster_kwargs={"image": "prefecthq/prefect", "n_workers": 5}
)

ecs_with_localdaskexecutor.py

from prefect import Flow
from prefect.storage import S3
from prefect.run_configs import ECSRun
from prefect.executors import LocalDaskExecutor

with Flow("example") as flow:
    ...

flow.storage = S3("my-flows")
flow.run_config = ECSRun(cpu="2 vcpu", memory="4 GB")  # increase resource limits
flow.executor = LocalDaskExecutor(num_workers=8)  # Use a local dask cluster

local_with_dask_cloudprovider.py

from prefect import Flow
from prefect.storage import S3
from prefect.run_configs import LocalRun
from prefect.executors import DaskExecutor

with Flow("example") as flow:
    ...

flow.storage = S3("my-flows")
flow.run_config = LocalRun()
flow.executor = DaskExecutor(
    cluster_class="dask_cloudprovider.aws.FargateCluster",
    cluster_kwargs={"image": "prefecthq/prefect", "n_workers": 5}
)