pyspark - filtrar - seleccionar una columna pandas

Unir dos marcos de datos, seleccionar todas las columnas de una y algunas columnas de la otra

Question

Digamos que tengo un marco de datos de chispa df1, con varias columnas (entre las cuales la columna ''id'') y un marco de datos df2 con dos columnas, ''id'' y ''otro''.

¿Hay alguna manera de replicar el siguiente comando?

sqlContext.sql("SELECT df1.*, df2.other FROM df1 JOIN df2 ON df1.id = df2.id")

¿Usando solo funciones de pyspark como join (), select () y similares?

Tengo que implementar esta unión en una función y no quiero ser forzado a tener sqlContext como parámetro de función.

¡Gracias!

Answer 1

Aquí hay una solución que no requiere un contexto SQL, pero mantiene los metadatos de un DataFrame.

a = sc.parallelize([[''a'', ''foo''], [''b'', ''hem''], [''c'', ''haw'']]).toDF([''a_id'', ''extra''])
b = sc.parallelize([[''p1'', ''a''], [''p2'', ''b''], [''p3'', ''c'']]).toDF(["other", "b_id"])
c = a.join(b, a.a_id == b.b_id)

Entonces, c.show() produce:

+----+-----+-----+----+
|a_id|extra|other|b_id|
+----+-----+-----+----+
| a| foo| p1| a|
| b| hem| p2| b|
| c| haw| p3| c|
+----+-----+-----+----+

Answer 2

Asterisco ( * ) funciona con alias. Ex:

from pyspark.sql.functions import *
df1 = df.alias(''df1'')
df2 = df.alias(''df2'')
df1.join(df2, df1.id == df2.id).select(''df1.*'')

Answer 3

No estoy seguro si es la forma más eficiente, pero esto funcionó para mí:

from pyspark.sql.functions import col
df1.alias(''a'').join(df2.alias(''b''),col(''b.id'') == col(''a.id'')).select([col(''a.''+xx) for xx in a.columns] + [col(''b.other1''),col(''b.other2'')])

El truco está en:

[col(''a.''+xx) for xx in a.columns] : all columns in a
[col(''b.other1''),col(''b.other2'')] : some columns of b

Answer 4

Puede hacer la unión y luego seleccionar las columnas deseadas https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=dataframe%20join#pyspark.sql.DataFrame.join

Answer 5

soltar duplicado b_id

c = a.join(b, a.a_id == b.b_id).drop(b.b_id)