python - 高效合并大量pyspark DataFrame

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)

这个想法是迭代地合并大小大致相同的对，直到留下一个结果。这是一个原型(prototype):

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