Pyspark Dataframes - Как сделать объединение, когда 3 столбца создают уникальный ключ

Question

В первом DF 3 столбца создают уникальный первичный ключ:

df1.select(concat(col("col1"), col("col2"), col("col3"))

Во втором DF 1 столбец является первичным ключом:

df2.select("col4")

Как объединить эти 2 кадра данных на первичном Ключи?

Я пытался с:

final_df = df1.join( df2, df1.select(concat(col("col1"), col("col2"), col("col3"))) == df2.select("col4") )

получить ошибку:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/lib/spark/python/pyspark/sql/dataframe.py", line 1037, in join
    assert isinstance(on[0], Column), "on should be Column or list of Column"
AssertionError: on should be Column or list of Column

Answer 1

Попробуйте это:

from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql.functions import *

spark = SparkSession.builder.appName('Stack_overflow').getOrCreate()

#DEFINING TWO DF's
df1schema = 'df1col1 int,df1col2 int,df1col3 int,df1col4 int'
df1 = spark.createDataFrame([[1,2,3,4],[2,3,4,5],[3,4,5,6],[4,5,6,7]],schema=df1schema)
df2schema = 'df2col1 int,df2col2 int'
df2 = spark.createDataFrame([[123,4],[234,5],[345,6],[456,7]],schema=df2schema)

#JOIN
df1.join(df2,concat(df1.df1col1,df1.df1col2,df1.df1col3) == df2.df2col1).show()

и вы получите:

df1:pyspark.sql.dataframe.DataFrame = [df1col1: integer, df1col2: integer ... 2 more fields]
df2:pyspark.sql.dataframe.DataFrame = [df2col1: integer, df2col2: integer]
+-------+-------+-------+-------+-------+-------+
|df1col1|df1col2|df1col3|df1col4|df2col1|df2col2|
+-------+-------+-------+-------+-------+-------+
|      4|      5|      6|      7|    456|      7|
|      2|      3|      4|      5|    234|      5|
|      3|      4|      5|      6|    345|      6|
|      1|      2|      3|      4|    123|      4|
+-------+-------+-------+-------+-------+-------+

Если вы хотите, чтобы столбец был конкатенен из трех столбцов, создайте withcolumn и отбросьте другие столбцы.

Answer 2

Это решение работает, но ему нужно создать новый столбец для объединения этих трех ключевых столбцов, а затем присоединиться к вновь созданному столбцу.

from pyspark.sql import functions 

#create sample dataframe
l = [(1,'lolena' , '2019.06.29'),(2, 'Anna', '2017.02.08'),(2, 'Niloo', '2010.20.19')]
df1 = spark.createDataFrame(l, ['student_id', 'name', 'timestamp'])
df1.show()

l2 = [(4,'mon','ir' , 'bokaee'),(3, 'ahm', 'ad', 'eslahi'),(1,'lol','ena' , 'eslahi')]
df2 = spark.createDataFrame(l2, ['student_id', 'name_p1', 'name_p2', 'lastname'])
df2.show()

df2 = df2.withColumn('name', functions.concat(functions.col('name_p1'), functions.col('name_p2')))
print("new df:")
df2.show()


inner_join = df1.join(df2, df1.name == df2.name )
print("inner join:\n")
inner_join.show()