如何加入两个 Spark DataFrame 并操作它们的 share 列？

Question

我有 2 个这样的 DataFrame：

+--+-----------+
|id|some_string|
+--+-----------+
| a|        foo|
| b|        bar|
| c|        egg|
| d|        fog|
+--+-----------+

还有这个：

+--+-----------+
|id|some_string|
+--+-----------+
| a|        hoi|
| b|        hei|
| c|        hai|
| e|        hui|
+--+-----------+

我想加入他们成为这样的人：

+--+-----------+
|id|some_string|
+--+-----------+
| a|     foohoi|
| b|     barhei|
| c|     egghai|
| d|        fog|
| e|        hui|
+--+-----------+

因此，第一个数据帧中的 some_string 列连接到第二个数据帧中的 some_string 列。如果我正在使用

df_join = df1.join(df2,on='id',how='outer')

它会返回

+--+-----------+-----------+
|id|some_string|some_string|
+--+-----------+-----------+
| a|        foo|        hoi|
| b|        bar|        hei|
| c|        egg|        hai|
| d|        fog|       null|
| e|       null|        hui|
+--+-----------+-----------+

有什么办法吗？

Answer 1

您需要使用when 才能实现正确的连接。除此之外，您使用outer join 的方式几乎是正确的。

您需要检查这两列中的任何一个是Null 或not Null，然后执行concatenation。

from pyspark.sql.functions import col, when, concat
df1 = sqlContext.createDataFrame([('a','foo'),('b','bar'),('c','egg'),('d','fog')],['id','some_string'])
df2 = sqlContext.createDataFrame([('a','hoi'),('b','hei'),('c','hai'),('e','hui')],['id','some_string'])
df_outer_join=df1.join(df2.withColumnRenamed('some_string','some_string_x'), ['id'], how='outer')
df_outer_join.show()
+---+-----------+-------------+
| id|some_string|some_string_x|
+---+-----------+-------------+
|  e|       null|          hui|
|  d|        fog|         null|
|  c|        egg|          hai|
|  b|        bar|          hei|
|  a|        foo|          hoi|
+---+-----------+-------------+
df_outer_join = df_outer_join.withColumn('some_string_concat',
                                         when(col('some_string').isNotNull() & col('some_string_x').isNotNull(),concat(col('some_string'),col('some_string_x')))
                                         .when(col('some_string').isNull() & col('some_string_x').isNotNull(),col('some_string_x'))
                                         .when(col('some_string').isNotNull() & col('some_string_x').isNull(),col('some_string')))\
                              .drop('some_string','some_string_x')


df_outer_join.show()
+---+------------------+
| id|some_string_concat|
+---+------------------+
|  e|               hui|
|  d|               fog|
|  c|            egghai|
|  b|            barhei|
|  a|            foohoi|
+---+------------------+

Answer 2

考虑到您想要执行外部联接，您可以尝试以下操作：

from pyspark.sql.functions import concat, col, lit, when


df_join= df1.join(df2,on='id',how='outer').when(isnull(df1.some_string1), ''). when(isnull(df2.some_string2),'').withColumn('new_column',concat(col('some_string1'),lit(''),col('some_string2'))).select('id','new_column')

（请注意，some_string1 和 2 指的是 df1 和 df2 数据帧中的 some_string 列。我建议您将它们命名为不同的名称，而不是使用相同的名称 some_string，以便您可以调用它们）