仅使用 Spark-SQL API 时广播变量的使用

Question

在使用 Spark-RDD API 时，我们可以使用广播变量来优化 spark 分配不可变状态的方式。

1) 广播变量如何在内部工作？

我的假设是： 对于每个用于对数据集执行操作的闭包，它所引用的所有变量都必须被序列化、通过网络传输并与任务一起恢复，以便可以执行闭包。

像这样注册广播变量时​​：

val broadcastVar = sc.broadcast("hello world")

返回的对象（Broadcast[String]）不保留对实际对象（“hello world”）的引用，而只保留一些 ID。 当一个广播变量句柄从上面所说的闭包中被引用时，它将像所有其他变量一样被序列化 - 只是广播变量句柄本身不包含实际对象。

稍后在目标节点上执行闭包时，实际对象（“hello world”）已经传输到每个节点。当闭包到达调用 broadcastVar.value 的位置时，广播变量句柄会在内部使用 ID 检索实际对象。

这个假设正确吗？

2) 有没有办法在 Spark-SQL 中利用这种机制？

假设我有一组允许的值。

当使用 RDD-API 时，我会为我的 allowedValues 创建一个广播变量：

val broadcastAllowedValues = sc.broadcast(allowedValues) // Broadcast[Set[String]]

rdd.filter(row => broadcastAllowedValues.value.contains(row("mycol")))

当然，在使用 Spark-SQL-API 时，我会使用 Column.isin / Column.isInCollection 方法：

dataframe.where(col("mycol").isInCollection(allowedValues))

但我似乎无法通过这种方式获得广播变量的优势。

另外，如果我将这段代码更改为以下内容：

val broadcastAllowedValues = sc.broadcast(allowedValues) // Broadcast[Set[String]]

dataframe.where(col("mycol").isInCollection(allowedValues.value))

这部分：

col("mycol").isInCollection(allowedValues.value)
// and more important this part:
allowedValues.value

将已在驱动程序上进行评估，从而产生一个新的Column-Object。所以广播变量在这里失去了它的优势。与第一个示例相比，它甚至会产生一些开销...

有没有办法使用 Spark-SQL-API 来利用广播变量，或者我必须在这些点上显式使用 RDD-API？

Answer 1

广播变量如何在内部工作？

广播的数据被序列化并物理移动到所有执行器。根据Broadcast Variables 上的文档，它说

“广播变量允许程序员在每台机器上缓存一个只读变量，而不是随任务一起发送它的副本。”

有没有办法在 Spark-SQL 中利用这种机制？

是的，有一种方法可以利用。 Spark 在加入大小数据帧时默认应用 Broadcast Hash Join。

根据“Learning Spark - 2nd edition”一书，它说：

"默认情况下，如果较小的数据集小于 10MB，Spark 将使用广播连接。此配置在spark.sql.autoBroadcastJoinThreshold 中设置；您可以根据每个执行器上的内存量和在司机。”

在您的情况下，您需要将所有唯一的 allowedValues 列出到一个只有一列（称为 allowValues 的列）的简单数据框（称为 allowedeValuesDF 的数据框）并应用连接来过滤您的 @ 987654328@.

类似这样的：

import org.apache.spark.sql.functions.broadcast
val result = dataframe.join(broadcast(allowedValuesDF), "mycol === allowedValues")

实际上，您可以省略 broadcast，因为 Spark 默认会进行广播加入。

编辑：

在更高版本的 Spark 中，您还可以在 SQL 语法中使用 join hints 来告诉执行引擎使用哪些策略。 SQL Documentation 中提供了详细信息，下面提供了一个示例：

-- Join Hints for broadcast join 
SELECT /*+ BROADCAST(t1) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key;