Spark Dataframe distinguir columnas con nombre duplicado

Comencemos con algunos datos:

from pyspark.mllib.linalg import SparseVector
from pyspark.sql import Row

df1 = sqlContext.createDataFrame([
    Row(a=107831, f=SparseVector(
        5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0})),
    Row(a=125231, f=SparseVector(
        5, {0: 0.0, 1: 0.0, 2: 0.0047, 3: 0.0, 4: 0.0043})),
])

df2 = sqlContext.createDataFrame([
    Row(a=107831, f=SparseVector(
        5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0})),
    Row(a=107831, f=SparseVector(
        5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0})),
])

Hay algunas maneras de abordar este problema. En primer lugar, puede hacer referencia inequívoca a las columnas de la tabla secundaria utilizando las columnas principales:

df1.join(df2, df1['a'] == df2['a']).select(df1['f']).show(2)

##  +--------------------+
##  |                   f|
##  +--------------------+
##  |(5,[0,1,2,3,4],[0...|
##  |(5,[0,1,2,3,4],[0...|
##  +--------------------+

También puede usar alias de tabla:

from pyspark.sql.functions import col

df1_a = df1.alias("df1_a")
df2_a = df2.alias("df2_a")

df1_a.join(df2_a, col('df1_a.a') == col('df2_a.a')).select('df1_a.f').show(2)

##  +--------------------+
##  |                   f|
##  +--------------------+
##  |(5,[0,1,2,3,4],[0...|
##  |(5,[0,1,2,3,4],[0...|
##  +--------------------+

Finalmente puede cambiar el nombre de las columnas mediante programación:

df1_r = df1.select(*(col(x).alias(x + '_df1') for x in df1.columns))
df2_r = df1.select(*(col(x).alias(x + '_df2') for x in df2.columns))

df1_r.join(df2_r, col('a_df1') == col('a_df2')).select(col('f_df1')).show(2)

## +--------------------+
## |               f_df1|
## +--------------------+
## |(5,[0,1,2,3,4],[0...|
## |(5,[0,1,2,3,4],[0...|
## +--------------------+

Después de indagar en la API de Spark, descubrí que primero puedo usar alias para crear un alias para el dataframe original y luego usar withColumnRename para cambiar el nombre manualmente de cada columna en el alias, al final para hacer el join sin causar la duplicación del nombre de la columna.

Se pueden consultar más detalles a continuación Spark Dataframe API :

Pyspark.SQL.DataFrame.alias

Pyspark.SQL.DataFrame.withColumnRenamed

Sin embargo, creo que esto es sólo un problema solución, y preguntándose si hay alguna mejor manera para mi pregunta.

Puede usar el método def drop(col: Column) para eliminar la columna duplicada, por ejemplo:

DataFrame:df1

+-------+-----+
| a     | f   |
+-------+-----+
|107831 | ... |
|107831 | ... |
+-------+-----+

DataFrame:df2

+-------+-----+
| a     | f   |
+-------+-----+
|107831 | ... |
|107831 | ... |
+-------+-----+

Cuando uno df1 con df2, el DataFrame será como a continuación:

val newDf = df1.join(df2,df1("a")===df2("a"))

DataFrame:newDf

+-------+-----+-------+-----+
| a     | f   | a     | f   |
+-------+-----+-------+-----+
|107831 | ... |107831 | ... |
|107831 | ... |107831 | ... |
+-------+-----+-------+-----+

Ahora, podemos usar el método def drop(col: Column) para eliminar la columna duplicada ' a 'o' f', de la siguiente manera:

val newDfWithoutDuplicate = df1.join(df2,df1("a")===df2("a")).drop(df2("a")).drop(df2("f"))

Hay una forma más sencilla que escribir alias para todas las columnas a las que te unes haciendo:

df1.join(df2,['a'])

Esto funciona si la clave con la que se está uniendo es la misma en ambas tablas.

Véase https://docs.databricks.com/spark/latest/faq/join-two-dataframes-duplicated-column.html

Así es como podemos unir dos Dataframes en los mismos nombres de columna en PySpark.

df = df1.join(df2, ['col1','col2','col3'])

Si lo hace printSchema() después de esto, puede ver que se han eliminado las columnas duplicadas.

Supongamos que los DataFrames que desea unir son df1 y df2, y los está uniendo en la columna 'a', entonces tiene 2 métodos

Método 1

Df1.join (df2,'a','left_outer')

Este es un método impresionante y es muy recomendable.

Método 2

Df1.unirse (df2, df1.a = = df2.a, 'left_outer').drop (df2.a)