python - Dask 图形执行和内存使用

Question

我正在 dask 中构建一个非常大的 DAG 以提交给分布式调度程序，其中节点在数据帧上运行，而数据帧本身可能非常大。一种模式是我有大约 50-60 个函数来加载数据并构建每个数百 MB 的 pandas 数据帧(并且在逻辑上代表单个表的分区)。我想将这些连接到图中下游节点的单个 dask 数据帧中，同时最小化数据移动。我这样链接任务:

dfs = [dask.delayed(load_pandas)(i) for i in disjoint_set_of_dfs]
dfs = [dask.delayed(pandas_to_dask)(df) for df in dfs]
return dask.delayed(concat_all)(dfs)

在哪里

def pandas_to_dask(df):
    return dask.dataframe.from_pandas(df).to_delayed()

我已经尝试了各种concat_all 实现，但这似乎是合理的:

def concat_all(dfs):
    dfs = [dask.dataframe.from_delayed(df) for df in dfs]
    return dask.dataframe.multi.concat(dfs, axis='index', join='inner')

所有 pandas 数据帧在其索引上都是不相交的，并且是排序/单调的。

然而，我正在死于这个 concat_all 函数的工作人员(集群管理器正在杀死他们超过他们的内存预算)即使每个人的内存预算实际上相当大而且我不会'不要指望它会四处移动数据。我有理由相信，在使用 dask 数据帧的图形节点中调用 compute() 之前，我总是会切片到合理的数据子集。

我正在玩 --memory-limit 到目前为止没有成功。我至少正确地解决了这个问题吗？有没有我遗漏的考虑因素？

Answer 1

给定计算到 pandas 数据帧的延迟值列表

>>> dfs = [dask.delayed(load_pandas)(i) for i in disjoint_set_of_dfs]
>>> type(dfs[0].compute())  # just checking that this is true
pandas.DataFrame

将它们传递给 dask.dataframe.from_delayed功能

>>> ddf = dd.from_delayed(dfs)

默认情况下，这将运行第一个计算以确定元数据(对 dask.dataframe 很重要的列名、数据类型等)。您可以通过构建示例数据框并将其传递给 meta= 关键字来避免这种情况。

>>> meta = pd.DataFrame({'value': [1.0], 'name': ['foo'], 'id': [0]})
>>> ddf = dd.from_delayed(dfs, meta=meta)

这example notebook也可能有帮助。

通常，您永远不需要从其他 dask 函数中调用 dask 函数(就像您通过延迟 from_pandas 调用所做的那样)。 Dask.dataframe 函数本身已经是惰性的，不需要进一步延迟。