python - 如何连接两个Spark DataFrame并操作它们的共享列？

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|
+---+------------------+