GitHub
06/29/2019, 6:45 PMJeremiah
06/29/2019, 7:30 PMZach Angell
06/29/2019, 7:51 PMwilsojb
07/01/2019, 7:28 PMJeremiah
07/01/2019, 10:55 PMJay Leach
07/02/2019, 4:23 AMChris White
07/02/2019, 5:12 AMJamie Catherwood
07/02/2019, 1:12 PMJeremiah
07/02/2019, 1:17 PMJamie Catherwood
07/02/2019, 1:18 PMhttps://media.tenor.com/images/02a2c2bd9eeafa9f15a1b072678ac59d/tenor.gif▾
Dylan Baker
07/02/2019, 3:16 PMChris Hart
07/02/2019, 6:19 PMChris White
07/02/2019, 9:55 PMRomain
07/03/2019, 12:25 PMAlistair Hey
07/05/2019, 12:28 PMJoselin
07/08/2019, 6:58 AMJoselin
07/08/2019, 6:58 AMChris Hart
07/08/2019, 7:00 PMChris Hart
07/08/2019, 7:02 PMDavid Norrish
07/09/2019, 12:17 AMJason
07/09/2019, 12:37 PMSherman
07/09/2019, 4:36 PMDavid Ojeda
07/10/2019, 3:58 PMprefect.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?Chris White
07/10/2019, 3:59 PMChris White
07/10/2019, 4:00 PMmap
method of the DaskExecutor
-> when we map tasks, we use executor.map
instead of executor.submit
David Ojeda
07/10/2019, 4:01 PMDaskExecutor.map
does not receive kwargs
, and I am using the “task” kwarg to guess the task name.Chris White
07/10/2019, 4:03 PMChris White
07/10/2019, 4:04 PMChris White
07/10/2019, 4:04 PMDaskExecutor.map
, try:
task_name = prefect.context.get("task_full_name", "some-default-value")
David Ojeda
07/10/2019, 4:13 PMDaskExecutor
map code, because I cannot do super().map(...)
given that the executor does not admit kwargs.
I can live with that, but I will have to keep an eye on any new changes on that class on future versions.David Ojeda
07/10/2019, 4:13 PMDaskExecutor
map code, because I cannot do super().map(...)
given that the executor does not admit kwargs.
I can live with that, but I will have to keep an eye on any new changes on that class on future versions.