高效合并大量 pyspark DataFrames

Question

我正在尝试在 Python 列表中执行数千个数据帧的数据帧联合。我正在使用我发现的两种方法。第一个使用for loop union，第二个使用functools.reduce。它们都适用于玩具示例，但是对于数千个数据帧，我正在试验严重的开销，这可能是由 JVM 之外的代码引起的，一次按顺序附加每个数据帧（使用两种合并方法）。

from functools import reduce  # For Python 3.x
from pyspark.sql import DataFrame

# The reduce approach
def unionAll(dfs):
    return reduce(DataFrame.unionAll, dfs)

df_list = [td2, td3, td4, td5, td6, td7, td8, td9, td10]
df = unionAll(df_list)

#The loop approach
df = df_list[0].union(df_list[1])
for d in df_list[2:]:
    df = df.union(d)

问题是如何有效地执行这种多数据帧操作，可能会规避由一个接一个合并数据帧引起的开销。

非常感谢

Answer 1

您目前正在像这样加入您的 DataFrame：

(((td1 + td2) + td3) + td4)

在每个阶段，您都将一个巨大的数据帧与一个小数据帧连接起来，从而导致每一步都有一个副本并浪费大量内存。我建议像这样组合它们：

(td1 + td2) + (td3 + td4)

这个想法是迭代合并大小大致相同的对，直到你得到一个结果。这是一个原型：

def pairwise_reduce(op, x):
    while len(x) > 1:
        v = [op(i, j) for i, j in zip(x[::2], x[1::2])]
        if len(x) > 1 and len(x) % 2 == 1:
            v[-1] = op(v[-1], x[-1])
        x = v
    return x[0]

result = pairwise_reduce(DataFrame.unionAll, df_list)

---

您将看到这对 python 列表有何巨大影响。

from functools import reduce 
from operator import add

x = [[1, 2, 3], [4, 5, 6], [7, 8], [9, 10, 11, 12]] * 1000

%timeit sum(x, [])
%timeit reduce(add, x)
%timeit pairwise_reduce(add, x)

64.2 ms ± 606 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)
66.3 ms ± 679 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)
970 µs ± 9.02 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)

sum(x, []) == reduce(add, x) == pairwise_reduce(add, x)
# True