Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Hi, is there a proper way to launch an on demand k8s dask cluster i.e. KubeCluster and then run prefect tasks on it using DaskExecutor? I tried doing it but getting a
```ERROR - prefect.FlowRunner | Unexpected error: TypeError('code() takes at most 15 arguments (16 given)')```
not sure if it’s a supported mode of operation.

Hi Tsang, yes it should work as you expect if you pass a KubeCluster to the Dask executor (see executor docs here: <https://docs.prefect.io/api/latest/engine/executors.html#daskexecutor|https://docs.prefect.io/api/latest/engine/executors.html#daskexecutor>).  What did you try?

yea so I got some simple tasks setup as ‘flow’ and then try to use this following syntax to run. was expecting some pods running but I ended up with a pickle error..
```dask.config.set({'kubernetes.worker_template_path': 'my-prefect-dask-worker.yml'})

dask_executor = DaskExecutor(
    cluster_class=KubeCluster,
    cluster_kwargs={
        'name': 'dask-worker-{JUPYTERHUB_USER}-{uuid}',
        'n_workers': 3
    },
)
flow.run(executor=dask_executor)```


the error trace back to
```Traceback (most recent call last):
  File "/opt/conda/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/opt/conda/lib/python3.8/site-packages/prefect/engine/flow_runner.py", line 600, in get_flow_run_state
    final_states = executor.wait(
  File "/opt/conda/lib/python3.8/site-packages/prefect/engine/executors/dask.py", line 451, in wait
    return self.client.gather(futures)
  File "/opt/conda/lib/python3.8/site-packages/distributed/client.py", line 1987, in gather
    return self.sync(
  File "/opt/conda/lib/python3.8/site-packages/distributed/client.py", line 833, in sync
    return sync(
  File "/opt/conda/lib/python3.8/site-packages/distributed/utils.py", line 339, in sync
    raise exc.with_traceback(tb)
  File "/opt/conda/lib/python3.8/site-packages/distributed/utils.py", line 323, in f
    result[0] = yield future
  File "/opt/conda/lib/python3.8/site-packages/tornado/gen.py", line 735, in run
    value = future.result()
  File "/opt/conda/lib/python3.8/site-packages/distributed/client.py", line 1852, in _gather
    raise exception.with_traceback(traceback)
  File "/usr/local/lib/python3.7/site-packages/distributed/protocol/pickle.py", line 75, in loads
TypeError: code() takes at most 15 arguments (16 given)```
perhaps I have a bad version?
```distributed==2.28.0
prefect==0.13.4```

Yea this seems like a dask / dask-kubernetes issue - perhaps search their repo for this error message?

Actually maybe try upgrading cloudpickle 

I’m already on
```cloudpickle==1.6.0```

(hope it’s ok to post the full trace) and thanks for the pointer. I’ll have to keep looking…

That's a serialization error, I suspect there's a version difference between your local process and the dask worker nodes. Things to check:
• python version
• cloud pickle version
• prefect version
• distributed version
The versions of these in the dask image your workers are running should match those you're using locally.

just recreated the environment with matching versions. it worked. thanks guys.