python - Dask 和字段名称中的多个合并

Question

我正在尝试将多个 pandas 数据帧合并到一个带有字段 ["a_id"、"b_id"、"c_id"] 的大型 Dask 数据帧上。每个 pandas 数据框“A”、“B”和“C”都有一个唯一字段(“a_id”、“b_id”和“c_id”)，将其连接到 Dask 数据框。 “B”和“C”也有一个字段“b_Field1”:

import pandas as pd
import dask.dataframe as dd

A = pd.DataFrame({'a_id': [1, 2, 3], 'a_Field1': [0, 0, 0]})
B = pd.DataFrame({'b_id': [3, 4, 5], 'b_Field1': [7, 8, 9]})
C = pd.DataFrame({'c_id': [4, 5, 6], 'b_Field1': [6, 7, 8], 'c_Field1': [10, 11, 12]})

pdf = pd.DataFrame({'a_id': [1, 2], 'b_id': [3, 4], 'c_id': [5, 6]})
pdf = pdf.merge(A, how="left", on="a_id")
pdf = pdf.merge(B, how="left", on="b_id")
pdf = pdf.merge(C, how="left", on=["c_id", "b_Field1"])

print(pdf)

"""
Result:
   a_id  b_id  c_id  a_Field1  b_Field1  c_Field1
0     1     3     5         0         7        11
1     2     4     6         0         8        12
"""

dA = dd.from_pandas(A, npartitions=1)
dB = dd.from_pandas(B, npartitions=1)
dC = dd.from_pandas(C, npartitions=1)
ddf = dd.from_pandas(pdf, npartitions=1)

ddf = ddf.merge(dA, how="left", on="a_id")
ddf = ddf.merge(dB, how="left", on="b_id")
ddf = ddf.merge(dC, how="left", on=["c_id", "b_Field1"])

这个失败了，说ddf中没有字段“b_Field1”。我的想法是我需要在合并 B 和 C 之间执行一个 .compute() 命令，但这会导致 Dask 在进度条上以 40% 的速度无休止地挂起(最终因 MemoryError 而死)。

在进行第二次连接之前是否需要进行计算？如果是这样，它会挂起的任何原因是什么？这些数据集刚好小到可以合并到纯 Pandas 中，而且合并发生得很快，但我正在努力使其可部署在内存较少的机器上。

Answer 1

如果您检查最后一行之前的数据框，您会发现它有列:

a_id  b_id  c_id  a_Field1_x  b_Field1_x  c_Field1  a_Field1_y  b_Field1_y

即，b_Field1 已一分为二，事实证明两者是相同的。这可能是 Dask 中的一个错误，因为正如您所展示的那样，同样的情况不会发生在 Pandas 中。但是，设置适当的索引或将可选参数调整为 merge 可能是一种解决方法。

有了数据框，你可以做

ddf = ddf.merge(dC.rename(columns={'b_Field1': 'b_Field1_x'}), 
     how="left", on=["c_id", "b_Field1_x"])

现在您还可以获得重复的 c_ 列。

关于一般内存问题，这在其他地方有很长的讨论。请务必谨慎选择您的分区大小、索引和工作器数量。