Scalanie wielu ramek danych w wierszach w PySpark

Mam 10 ramek danych pyspark.sql.dataframe.DataFrame, uzyskanych od randomSplitjak (td1, td2, td3, td4, td5, td6, td7, td8, td9, td10) = td.randomSplit([.1, .1, .1, .1, .1, .1, .1, .1, .1, .1], seed = 100)Teraz chcę dołączyć 9 td„s na pojedynczej ramce danych, jak mam to zrobić?

Próbowałem już z unionAll, ale ta funkcja akceptuje tylko dwa argumenty.

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)

Czy istnieje sposób na połączenie więcej niż dwóch ramek danych w rzędzie?

Celem tego jest to, że robię 10-krotną weryfikację krzyżową ręcznie bez użycia CrossValidatormetody PySpark , więc biorąc 9 na trening i 1 na dane testowe, a następnie powtórzę to dla innych kombinacji.

Skradzione z: /programming/33743978/spark-union-of-multiple-rdds

Poza łańcuchami związków jest to jedyny sposób, aby to zrobić dla 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)

Dzieje się tak, ponieważ bierze wszystkie przekazane obiekty jako parametry i redukuje je za pomocą unionAll (ta redukcja pochodzi z Pythona, a nie redukcji Spark, chociaż działają podobnie), co ostatecznie redukuje ją do jednej ramki danych.

Jeśli zamiast DataFrames są one normalnymi RDD, możesz przekazać ich listę do funkcji unii SparkContext

EDYCJA: Dla twojego celu proponuję inną metodę, ponieważ musiałbyś powtórzyć ten cały związek 10 razy dla różnych fałd dla krzyżowej walidacji, dodałbym etykiety, do których należy fałd, i po prostu filtruję twoją ramkę danych dla każdego fałdu na podstawie etykieta

Czasami, gdy łączone ramki danych nie mają tej samej kolejności kolumn, lepiej jest df2.select (df1.columns), aby upewnić się, że oba df mają tę samą kolejność kolumn przed zjednoczeniem.

import functools 

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

Przykład:

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

w przeciwnym razie wygeneruje poniższy wynik.

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

Co powiesz na użycie rekurencji?

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])