apache-spark - 为什么 Spark 在 joinWith 期间对预先分区的数据帧执行不必要的随机播放？

Question

此示例已使用 Spark 2.4.x 进行了测试。让我们考虑 2 个简单的数据框:

case class Event(user_id: String, v: String)
case class User(user_id: String, name: String)
val events = spark.createDataFrame(Seq(Event("u1", "u1.1"),Event("u1", "u1.2"),Event("u2", "u2.1")))
val users = spark.createDataFrame(Seq(User("u1", "name1"),User("u2", "name2"),User("u3", "name3")))

当在同一列上有 groupBy 后跟 joinWith 时，Spark 会执行不必​​要的洗牌:

events
  .groupBy("user_id").agg(collect_list(struct("v")).alias("values"))
  .joinWith(users, events("user_id") === users("user_id"), "full_outer")
  .show

当用连接做同样的事情时，额外的洗牌消失了:

events
  .groupBy("user_id").agg(collect_list(struct("v")).alias("values"))
  .join(users, events("user_id") === users("user_id"), "full_outer")
  .show

这是一个已知问题吗？是否有使用 joinWith 而无需额外随机播放的解决方法？

附录:这里是 SQL 计划

• 与 joinWith

• 与加入

Answer 1

这似乎是由 this ticket 中的错误修复引入的错误. 外连接 的结果是错误的。因此需要在 Join 节点之前添加一个 Project 节点(结构的打包)。

但是，我们最终得到了这种查询计划:

Join FullOuter, (_1#122.user_id = _2#123.user_id)
:- Project [named_struct(user_id, user_id#4, values, values#115) AS _1#122]
:  +- Aggregate [user_id#4], [user_id#4, collect_list(struct(v, v#5), 0, 0) AS values#115]
:     +- LocalRelation [user_id#4, v#5]
+- LocalRelation [_2#123]

Project 节点阻止 catalyst 优化额外的 shuffle。

但是，对于 this PR Project 节点已被删除以进行内部连接，因此需要进行额外的随机播放。