dask - 在 Dask 数据帧子集上强制局部性

Question

我正在尝试在多台机器上分发一个大型 Dask 数据帧，以便(稍后)在数据帧上进行分布式计算。我为此使用了 dask-distributed。

我看到的所有 dask 分布式示例/文档都是从网络资源(hdfs、s3 等)填充初始数据负载，并且似乎没有将 DAG 优化扩展到负载部分(似乎假设网络负载是必要的邪恶，只是吃掉初始成本。)这在另一个问题的答案中得到了强调:Does Dask communicate with HDFS to optimize for data locality?

但是，我可以看到我们想要这个的情况。例如，如果我们在这个数据库的节点上有一个分片数据库 + dask 工作人员，我们希望只将本地分片中的记录填充到本地 dask 工作人员中。从文档/示例来看，网络交叉似乎是一个必然的假设成本。 是否可以强制从特定工作人员获取单个数据帧的部分内容？

我尝试过的另一种方法是尝试强制每个工作人员运行一个函数(迭代提交给每个工作人员)，其中该函数仅加载该机器/分片的本地数据。这有效，并且我有一堆具有相同列模式的最佳本地数据帧 - 但是 - 现在我没有单个数据帧而是 n 个数据帧。 是否可以跨多台机器合并/融合数据帧，以便有一个单一的数据帧引用，但部分与特定机器具有关联性(在合理范围内，由任务 DAG 决定)？

Answer 1

您可以生成 dask “集合”，例如来自 future 和延迟对象的数据帧，它们之间可以很好地互操作。

对于每个分区，您知道哪台机器应该加载它，您可以按如下方式生成一个 future :

f = c.submit(make_part_function, args, workers={'my.worker.ip'})

哪里 c是 dask 客户端，地址是您希望看到它发生的机器。您也可以给 allow_other_workers=True这是一种偏好而不是要求。

要从此类 future 列表中制作数据框，您可以执行以下操作

df = dd.from_delayed([dask.delayed(f) for f in futures])

最好提供一个 meta= ，给出预期数据帧的描述。现在，对给定分区的进一步操作将更喜欢在已经保存数据的同一工作人员上安排。