Pyspark Dataframes - 如何在 3 列成为唯一键的情况下进行连接

Question

在第一个 DF 中，3 列构成唯一主键：

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

在第二个 DF 中，1 列是主键：

df2.select("col4")

如何在主键上加入这两个数据框？

我试过了：

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