Объединение нескольких фреймов данных построчно в PySpark

У меня есть 10 фреймов данных pyspark.sql.dataframe.DataFrame, полученных randomSplitкак. (td1, td2, td3, td4, td5, td6, td7, td8, td9, td10) = td.randomSplit([.1, .1, .1, .1, .1, .1, .1, .1, .1, .1], seed = 100)Теперь я хочу объединить 9 tdфреймов в один фрейм данных, как мне это сделать?

Я уже пробовал с unionAll, но эта функция принимает только два аргумента.

td1_2 = td1.unionAll(td2) 
# this is working fine

td1_2_3 = td1.unionAll(td2, td3) 
# error TypeError: unionAll() takes exactly 2 arguments (3 given)

Есть ли способ объединить более двух кадров данных по строкам?

Цель этого состоит в том, что я делаю 10-кратную перекрестную валидацию вручную без использования CrossValidatorметода PySpark , поэтому я беру 9 в тренировку и 1 в тестовые данные, а затем я повторю это для других комбинаций.

Похищен у: /programming/33743978/spark-union-of-multiple-rdds

Вне цепочки союзов это единственный способ сделать это для DataFrames.

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

def unionAll(*dfs):
    return reduce(DataFrame.unionAll, dfs)

unionAll(td2, td3, td4, td5, td6, td7, td8, td9, td10)

Что происходит, так это то, что он принимает все объекты, которые вы передали в качестве параметров, и уменьшает их с помощью unionAll (это сокращение от Python, а не от Spark, хотя они работают аналогично), которое в конечном итоге сводит его к одному DataFrame.

Если вместо DataFrames они являются обычными RDD, вы можете передать их список в функцию объединения вашего SparkContext.

РЕДАКТИРОВАТЬ: Для вашей цели я предлагаю другой метод, так как вам нужно будет повторить весь этот союз 10 раз для ваших разных сгибов для перекрестной проверки, я бы добавил метки, которым принадлежит сгиб строки и просто отфильтровать ваш DataFrame для каждого сгиба на основе наклейка

Иногда, когда объединяемые кадры данных не имеют одинаковый порядок столбцов, лучше использовать df2.select (df1.columns), чтобы оба объединения имели одинаковый порядок столбцов перед объединением.

import functools 

def unionAll(dfs):
    return functools.reduce(lambda df1,df2: df1.union(df2.select(df1.columns)), dfs) 

Пример:

df1 = spark.createDataFrame([[1,1],[2,2]],['a','b'])
# different column order. 
df2 = spark.createDataFrame([[3,333],[4,444]],['b','a']) 
df3 = spark.createDataFrame([555,5],[666,6]],['b','a']) 

unioned_df = unionAll([df1, df2, df3])
unioned_df.show() 

иначе это произведет приведенный ниже результат.

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

def unionAll(*dfs):
    return reduce(DataFrame.unionAll, dfs) 

unionAll(*[df1, df2, df3]).show()

Как насчет использования рекурсии?

def union_all(dfs):
    if len(dfs) > 1:
        return dfs[0].unionAll(union_all(dfs[1:]))
    else:
        return dfs[0]

td = union_all([td1, td2, td3, td4, td5, td6, td7, td8, td9, td10])