I want to ask the Dask Experts whether they can help identify a scheduler bottleneck I have experienced. I have a flow that schedules 22 parallel mapped jobs (each with 65 items) allowing for 1430 parallel jobs. When I schedule the job on an k8s cluster with 375 nodes (each with a worker that runs 4 threads), my scheduler throws messages out like the below. The CPU utilization of the scheduler also goes to 100%. These jobs ultimately die when all the workers get disconnected.

distributed.utils_perf - WARNING - full garbage collections took 10% CPU time recently (threshold: 10%)
...
distributed.core - INFO - Event loop was unresponsive in Scheduler for 7.07s.  This is often caused by long-running GIL-holding functions or moving large chunks of data. This can cause timeouts and instability.
...
distributed.comm.tcp - INFO - Connection closed before handshake completed
...
distributed.scheduler - INFO - Close client connection: Client-worker-3c6d8642-53b5-11eb-800e-32b98c347770

When I scale the job down to 250 nodes and 3 threads per worker, I still get 100% utilization, but it is slightly more stable. where the only warning messages is

distributed.utils_perf - WARNING - full garbage collections took 11% CPU time recently (threshold: 10%)

Thanks for any advice!

did you try increasing the resources (CPU and memory) allocated for the scheduler ?

I set no limits on the scheduler. It seems to plateau at 1. Same with memory - no limits.

Hi @Vincent! Would you mind sharing your cluster configuration & Flow run config + storage?

Hi @Dylan I am running k8s on AWS using the DaskKubernetesEnvironment (from prefect 0.13.17) Running with DockerStorage. What configs are you interested in particular?

I believe that users DaskKubernetes under the hood: https://kubernetes.dask.org/en/latest/kubecluster.html

@Vincent you'll want to increase the resource spec by passing in a custom

scheduler_spec_file

to the

DaskKubernetesEnvironment

Thanks - I am aware of the changes, but as you mentioned, it still uses DaskKubernestes under the hood (with the KubeCluster) . I have not found any ways to get the scheduler to use more than 1 cpu.

this https://github.com/PrefectHQ/prefect/blob/master/src/prefect/environments/execution/dask/job.yaml is the default spec file that is being used with 100mcpu as a default

☝️ Marwan is right on the money 💯

Thanks @Marwan Sarieddine I set this to 1000 and removed the limits. It should be capable to scale past 1 but it does not.

hmm strange - I don't think it is

Just as a followup - I upgrade my setup to use 0.14.2 for the submission. Overall, I have noticed increased stability, when spawning the tasks. However, the scheduler is killed (with no error message) after ~ 10 minutes. =(

I see - perhaps someone from the prefect team can shed some more light here - sorry out of ideas ...

Yes that is correct - The job exits on the k8s cluster, and the job continues to run in the cloud service.

nothing in the kubernetes logs of the scheduler pod that point to any errors ? if you are using a cluster autoscaler make sure to annotate the scheduler and worker pods as not safe to evict

Thanks for the advice! I think it may have been an error on my side where I did not annotate the scheduler with any memory requests. I assumed that leaving it out would give it "unlimited" memory but I guess it just allowed the k8s cluster to move it as it wished.

glad to know you got it fixed