Dask分布式调度程序和大型函数

Question

在带有LocalCluster的Dask分布式调度程序的上下文中：有人可以帮助我了解具有大型（堆）映射功能的动态特性吗？

例如，考虑Dask Data Frame ddf和map_partitions操作：

def mapper():
  resource=... #load some large resource eg 50MB

  def inner(pdf):
    return pdf.apply(lambda x: ..., axis=1)

  return inner

mapper_fn = mapper() #50MB on heap
ddf.map_partitions(mapper_fn)

这里会发生什么？ Dask会序列化mapper_fn并发送给所有任务吗？说，我有n个分区，所以有n个任务。

[根据经验，我观察到，如果我有40个任务和一个50MB的映射器，则大约需要70 s taks才能开始工作，该集群似乎在这里坐满了CPU，但仪表板什么也没显示。这是怎么回事在碟形分布式调度程序中具有较大（堆）功能的结果是什么？

Answer 1

Dask使用cloudpickle序列化非平凡的功能，并在每个任务中包括这些功能的序列化版本。这是非常低效的。我们建议您不要这样做，而应显式传递数据。

resource = ...

ddf.map_partitions(func, resource=resource)

这将效率更高。