Dask bag 分区和工人如何关联？

Question

我正在使用带有两个工作人员和一个调度程序的 vanilla Dask-Kubernetes 设置来迭代一些 JSON 文件的行（并应用一些为简单起见未出现在此处的函数）。我只看到一个工人在工作，而我希望看到其中两个。

希望重新分区会有所帮助，我已经尝试了 bag.repartition(num) 的不同值，它们返回不同数量的行，但它们不会改变工作人员不平衡和仅集中在一个工作人员身上的内存消耗。

我想我不了解分区和工作人员之间的相关性，我在 Dask 文档中找不到任何关于它的内容。非常欢迎任何帮助或指点！

import dask.bag as db

def grep_buildings():
    base = "https://usbuildingdata.blob.core.windows.net/usbuildings-v1-1/"
    b = db.text.read_text(f"{base}/Alabama.zip")
    # b = b.repartition(2)
    lines = b.take(3_000_000)
    return lines

len(grep_buildings())

Answer 1

在您的示例中，您正在打开文件，并且它已被压缩

db.text.read_text(f"{base}/Alabama.zip")

Dask 能够并行打开和处理多个文件，每个文件至少有一个分区。 Dask 还能够将单个文件拆分为块（blocksize 参数）；但这只适用于未压缩的数据。原因是，对于全文件压缩方法，到达未压缩流中某个点的唯一方法是从头开始读取，因此每个分区都会读取大部分数据。

最后，当您从单个分区开始时，重新分区对您没有帮助：您需要先读取整个文件，然后再将数据拆分为下游任务。