Fusionar múltiples marcos de datos en fila en PySpark

Tengo 10 marcos de datos pyspark.sql.dataframe.DataFrame, obtenidos de randomSplitcomo (td1, td2, td3, td4, td5, td6, td7, td8, td9, td10) = td.randomSplit([.1, .1, .1, .1, .1, .1, .1, .1, .1, .1], seed = 100)Ahora quiero unir 9 tden un solo marco de datos, ¿cómo debo hacer eso?

Ya lo he intentado unionAll, pero esta función solo acepta dos argumentos.

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)

¿Hay alguna forma de combinar más de dos marcos de datos en fila?

El propósito de hacer esto es que estoy haciendo la validación cruzada 10 veces manualmente sin usar el CrossValidatormétodo PySpark , así que tomo 9 en el entrenamiento y 1 en los datos de prueba y luego lo repetiré para otras combinaciones.

Robado de: /programming/33743978/spark-union-of-multiple-rdds

Fuera de encadenar sindicatos, esta es la única forma de hacerlo para 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)

Lo que sucede es que toma todos los objetos que pasó como parámetros y los reduce usando unionAll (esta reducción es de Python, no la reducción de Spark aunque funcionan de manera similar), lo que finalmente lo reduce a un DataFrame.

Si en lugar de DataFrames son RDD normales, puede pasar una lista de ellos a la función de unión de su SparkContext

EDITAR: para su propósito, propongo un método diferente, ya que tendría que repetir esta unión completa 10 veces para sus diferentes pliegues para la validación cruzada, agregaría etiquetas para qué pliegue pertenece una fila y simplemente filtrar su DataFrame para cada pliegue según la etiqueta

En algún momento, cuando los marcos de datos para combinar no tienen el mismo orden de columnas, es mejor df2.select (df1.columns) para asegurar que ambos df tengan el mismo orden de columnas antes de la unión.

import functools 

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

Ejemplo:

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

de lo contrario, generaría el siguiente resultado en su lugar.

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

¿Qué tal usar la recursividad?

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