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

    # Create Dask Executor

    def make_cluster(
        image=None,
    ):
        """Start a cluster using the same image as the flow run"""

        pod_spec = make_pod_spec(
            image=image or prefect.context.image,  # default to current active image
            memory_limit=memory_limit,
            memory_request=memory_request,
            cpu_limit=cpu_limit,
            cpu_request=cpu_request,
        )
        return KubeCluster(
            pod_spec,
        )

    flow.executor = DaskExecutor(
        cluster_class=make_cluster, adapt_kwargs={"minimum": 0, "maximum": n_workers}
    )