apache-spark - 我可以在 sparkConf 中为不同的 sql 设置不同的 autoBroadcastJoinThreshold 值吗？

Question

我有大数据帧: A(200g), B(20m), C(15m), D(10m), E(12m) ，我想加入他们: A 加入 B , C 加入 D 和 E 在同一个 SparkSession** 中使用 spark sql。就像:

absql:sql("select * from A a inner join B b on a.id=b.id").write.csv("/path/for/ab")
cdesql:sql("select * from C c inner join D d on c.id=d.id inner join E e on c.id=e.id").write.csv("/path/for/cde")

问题:

当我使用默认 spark.sql.autoBroadcastJoinThreshold=10m

absql 需要很长时间，原因是 absql skew。

cdesql 正常

当我设置 spark.sql.autoBroadcastJoinThreshold=20m

C,D,E 将被广播，所有任务将在同一个执行器中执行，但仍然需要很长时间。

如果设置 num-executors=200，则广播时间较长

absql 正常

Answer 1

而不是改变 autoBroadcastJoinThreshold ，您可以标记要广播的数据帧。通过这种方式，很容易决定应该广播或不广播哪些数据帧。

在 Scala 中，它看起来像这样:

import org.apache.spark.sql.functions.broadcast
val B2 = broadcast(B)
B2.createOrReplaceTempView("B")

这里数据帧 B 已经被标记为广播，然后被注册为一个表以与 Spark SQL 一起使用。

或者，这可以直接使用数据帧 API 完成，第一个连接可以写为:

A.join(broadcast(B), Seq("id"), "inner")