如何正确加入 Apache Spark 中的 2 个数据框？答案

【问题标题】：How correctly to join 2 dataframe in Apache Spark?如何正确加入 Apache Spark 中的 2 个数据框？
【发布时间】：2019-05-02 05:27:02
【问题描述】：

我是 Apache Spark 的新手，需要一些帮助。有人能说如何正确地加入接下来的 2 个数据框吗？！

第一个数据帧：

| DATE_TIME           | PHONE_NUMBER |
|---------------------|--------------|
| 2019-01-01 00:00:00 | 7056589658   |
| 2019-02-02 00:00:00 | 7778965896   |

第二个数据框：

| DATE_TIME           | IP            |
|---------------------|---------------|
| 2019-01-01 01:00:00 | 194.67.45.126 |
| 2019-02-02 00:00:00 | 102.85.62.100 |
| 2019-03-03 03:00:00 | 102.85.62.100 |

我想要的最终数据框：

| DATE_TIME           | PHONE_NUMBER | IP            |
|---------------------|--------------|---------------|
| 2019-01-01 00:00:00 | 7056589658   |               |
| 2019-01-01 01:00:00 |              | 194.67.45.126 |
| 2019-02-02 00:00:00 | 7778965896   | 102.85.62.100 |
| 2019-03-03 03:00:00 |              | 102.85.62.100 |

下面是我尝试过的代码：

import org.apache.spark.sql.Dataset
import spark.implicits._

val df1 = Seq(
    ("2019-01-01 00:00:00", "7056589658"),
    ("2019-02-02 00:00:00", "7778965896")
).toDF("DATE_TIME", "PHONE_NUMBER")

df1.show()

val df2 = Seq(
    ("2019-01-01 01:00:00", "194.67.45.126"),
    ("2019-02-02 00:00:00", "102.85.62.100"),
    ("2019-03-03 03:00:00", "102.85.62.100")
).toDF("DATE_TIME", "IP")

df2.show()

val total = df1.join(df2, Seq("DATE_TIME"), "left_outer")

total.show()

不幸的是，它引发了错误：

org.apache.spark.SparkException: Exception thrown in awaitResult:
  at org.apache.spark.util.ThreadUtils$.awaitResult(ThreadUtils.scala:205)
  at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec.doExecuteBroadcast(BroadcastExchangeExec.scala:136)
  at org.apache.spark.sql.execution.InputAdapter.doExecuteBroadcast(WholeStageCodegenExec.scala:367)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeBroadcast$1.apply(SparkPlan.scala:144)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeBroadcast$1.apply(SparkPlan.scala:140)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
  at org.apache.spark.sql.execution.SparkPlan.executeBroadcast(SparkPlan.scala:140)
  at org.apache.spark.sql.execution.joins.BroadcastHashJoinExec.prepareBroadcast(BroadcastHashJoinExec.scala:135)
...

【问题讨论】：

这对我来说似乎是正确的，只是连接类型需要为 full 才能获得所需的结果。也许问题出在配置中，您能否发布完整的堆栈跟踪？

标签： scala apache-spark apache-spark-sql

【解决方案1】：

你需要full outer join，但你的代码很好。您的问题可能是其他问题，但是您提到的堆栈跟踪无法得出问题所在。

val total = df1.join(df2, Seq("DATE_TIME"), "full_outer")

【讨论】：

我在 Zeppelin 中测试了代码。我找到了错误原因。就我而言，我在开始时选择了不正确的interpreter binding。 full_outer 正是我需要的。谢谢你的回答。

【解决方案2】：

你可以这样做：

val total = df1.join(df2, (df1("DATE_TIME") === df2("DATE_TIME")), "left_outer")

【讨论】：