When running with a k8s agent, each flow run will create a new k8s job, which will have some delay before startup. There's no way around this, if you need faster startup times than k8s can start jobs you'll want to either use a different agent, or make your flow runs less fine grained so you don't need to run as many of them. That said, if your flows make use a large (distributed) dask cluster on k8s, you could speed up execution time by keeping a dask cluster running in the background that your flows connect to. You can do this by providing an address to a

DaskExecutor

configured on a flow:

flow.executor = DaskExecutor(address=...)

You might make use the helm chart (https://github.com/dask/helm-chart) for managing this.

Is it possible to avoid this overhead (or at least only pay it once), by telling the KubernetesAgent and/or DaskKubernetesEnvironment to always keep X worker pods alive which continue to wait for and complete tasks without dying until the end of the flow?

Dask shouldn't be starting a new worker for every task (the flow should startup a dask cluster, run all tasks on it, then shutdown the cluster). If you're using adaptive scaling the cluster might scale up/down during a flow run to add workers when needed and remove them when unneeded. There are some config options for how long a worker can sit idle before scaling down, if your flows have large gaps between work this may be triggering this behavior. You can also run with a fixed set of workers instead (pass

n_workers

to

cluster_kwargs

in the

DaskExecutor

constructor), which will disable adaptive scaling.

flow.executor = DaskExecutor(
    cluster_class=...,
    cluster_kwargs={"n_workers": 5, ...},
)

Also note that if you only need a small amount of parallelism, you may be better served with using a

LocalDaskExecutor

, which avoids all the complications of running multiple pods for a flow.

To give some more context about the flow: One of our flows currently consists of ~20 smaller tasks of which some are taking much longer than others. Other flows rather have 3-5 bigger tasks. All flows are executed for 10s of thousands of systems, hence high parallelization would be great (1000s to 10_000s parallel tasks ideally).

As Sev described above, we thought of batching together tasks that execute fast, but seem to have quite some overhead. But your explanations suggest that prefect should normally not have such a big overhead and that batching tasks should not be required (would also seem unpythonic/unintuitive). We would be happy to optimize our prefect setup and not need to think about workarounds or batching etc. 🙂

Hello again @Jim Crist-Harif, I have an update for you: I've set up a run_config with KubernetesRun, and an executer with DaskExecutor (using a KuberCluster). This is what you had in mind, I hope? Feedback here would be highly appreciated 🙂

# Create run config
flow.run_config = KubernetesRun(
    cpu_request=2, memory_request="2G", env={"AWS_DEFAULT_REGION": "eu-central-1"}
)


# Create Dask Executor
def make_cluster(n_workers, image):
    """Start a cluster using the same image as the flow run"""
    from dask_kubernetes import KubeCluster, make_pod_spec

    pod_spec = make_pod_spec(
        image=image,
        memory_limit="1900M",
        memory_request="1900M",
        cpu_limit=0.5,
        cpu_request=0.5,
    )

    return KubeCluster(pod_spec, n_workers=n_workers)


flow.executor = DaskExecutor(
    cluster_class=make_cluster,
    cluster_kwargs={"n_workers": 10, "image": flow.storage.name},
)

Anyway, in the end it behaved exactly as you suggested it would 😄. As in, when mapping over a task with 100 iterations, and

n_workers=5

, then I saw only 5 dask worker pods which worked throughout the run (very efficient!) If you're interested, you can see a full codebase of what I've written here, which will hopefully be helpful to others who are interested in a similar set-up. We're also plan to turn this in a blog post sometime soon(ish), also with information about how our team has mixed Prefect with Pulumi to unleash a robot army 🤖

In any case, thanks again for the helpful feedback. And also, props to the Prefect team for solving the parallelization problem with the newest version before we had even asked our question!

One small tip - you can access the current docker image used by the flow run at runtime via the

prefect.context.image

field. This would let you avoid passing in an image via

cluster_kwargs

to

DaskExecutor

, and let your

make_cluster

function determine the image automatically.

def make_cluster(n_workers, image=None):
    """Start a cluster using the same image as the flow run"""
    from dask_kubernetes import KubeCluster, make_pod_spec

    pod_spec = make_pod_spec(
        image=image or prefect.context.image,  # default to current active image
        memory_limit="1900M",
        memory_request="1900M",
        cpu_limit=0.5,
        cpu_request=0.5,
    )

    return KubeCluster(pod_spec, n_workers=n_workers)

See https://docs.prefect.io/orchestration/flow_config/executors.html#using-a-temporary-cluster for more info.