在 Pyspark 中合并 2 个数据框答案

【问题标题】：Merging 2 dataframe in Pyspark在 Pyspark 中合并 2 个数据框
【发布时间】：2018-10-31 18:21:03
【问题描述】：

我有 2 个数据框，如下所示：

DF1：

Id | field_A | field_B | field_C | field_D
 1 |   cat   |  12     |   black | 11
 2 |   dog   | 128     |   white | 19
 3 |   dog   |  35     |  yellow | 20
 4 |   dog   |  21     |   brown |  4
 5 |  bird   |  10     |    blue |  7
 6 |   cow   |  99     |   brown | 34

DF2：

Id | field_B | field_C | field_D | field_E
 3 |  35     |  yellow | 20      |   123   
 5 |  10     |    blue |  7      |   454   
 6 |  99     |   brown | 34      |   398

左合并后，我希望得到以下数据框：

Id | field_A | field_B | field_C | field_D | field_E
 1 |   cat   |  12     |   black | 11      |
 2 |   dog   | 128     |   white | 19      |
 3 |   dog   |  35     |  yellow | 20      |  123
 4 |   dog   |  21     |   brown |  4      |  
 5 |  bird   |  10     |    blue |  7      |  454
 6 |   cow   |  99     |   brown | 34      |  398

但我得到以下数据框：

Id | field_A | field_B | field_C | field_D | field_E 
 3 |   dog   |  35     |  yellow | 20      |  123
 5 |  bird   |  10     |    blue |  7      |  454
 6 |   cow   |  99     |   brown | 34      |  398

我正在使用以下语法：

new_df = df1.join(df2, on=['field_B', 'field_C', 'field_D'], how='left_outer')

我正在开发 Spark 2.2。谁能告诉我为什么会这样？谢谢！

【问题讨论】：

df 名称有没有可能是错误的？您的语法看起来正确。试试这个 new_df = df2.join(df1, on=['field_B', 'field_C', 'field_D'], how='left_outer')
感谢阿里！我尝试了所有 outer 连接的 linds。仍然得到相同的答案。并且变量名是正确的，否则会引发错误。
您能否确认 df1 和 df2 的顺序是否与您在问题中所说的相同？乍一看，似乎顺序颠倒了。要解决此订单问题，请尝试传递 how='outer'。如果可行，只需更改语句中 df1 和 df2 的顺序即可。
列的顺序在 spark 中并不重要，您可以随时将它们更改为您想要的任何内容。在您的情况下，只需在 join 之后执行 new_df = new_df.select("id", *["field_"+c for c in ["A", "B", "C", "D", "E"]])。

标签： pyspark

【解决方案1】：

您分享的代码 sn-p 完全可以正常工作。看起来加入的顺序与您在问题中所说的不同。试试
df2.join(df1, on='field_B', 'field_C', 'field_D'], how='left')

如果可行，您的数据框顺序与您在此处显示的不同。

【讨论】：

【解决方案2】：

根据这个答案，您必须使用下面链接中的unionDFs 函数，然后使用groupby。 https://stackoverflow.com/a/50665187/6645617

这是一个示例代码：

from pyspark.sql.functions import first

unionDF = unionDFs(DF1,DF2)
firstCols = [first(col).alias(str(col)) for col in unionDF.columns[1:]]

finalDF = unionDF.groupby("Id").agg(*firstCols)

【讨论】：

请查看我的答案并告诉我它是否适合您。