Эффективное объединение большого количества pyspark DataFrames

Question

Я пытаюсь выполнить объединение тысяч кадров в списке Python. Я использую два подхода, которые нашел. Первый - с помощью для объединения петель , а второй - 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

В настоящее время вы присоединяетесь к своим фреймам данных следующим образом:

(((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