Hi there, I was trying to get the dask scheduler UI to be a bit more expressive. Normally, I have all my running tasks grouped under the run_task and run_fn names, but what I would like is to have the real task name. I managed to extend

prefect.engine.executors.DaskExecutor

to add the task name as the

key

parameter sent to the dask client:

import uuid
from typing import Any, Callable

from distributed import Future
from prefect.engine.executors import DaskExecutor


class MyDaskExecutor(DaskExecutor):

    def submit(self, fn: Callable, *args: Any, **kwargs: Any) -> Future:
        if 'key' not in kwargs and 'task' in kwargs:
            kwargs['key'] = kwargs['task'].__class__.__name__ + '-' + str(uuid.uuid4())
        return super().submit(fn, *args, **kwargs)

So far this is working as expected, with the exception of tasks that use the

.map(...)

idiom, where the run_fn still appears with that name. I have looked further into prefect’s code but it seems that to achieve this I would need to create my own task runner. Since that seems a bit complicated, I just wanted to ask if there is another solution or if this is just a bad idea?