代码之家 › 专栏 › 技术社区 › vikrant rana

广播哈希联接

apache-spark-sql pyspark apache-spark

2

vikrant rana · 技术社区 · 7 年前

当我们有一个足够小的数据帧可以放入内存时,我们使用spark中的广播散列连接。当小数据帧的大小低于 spark.sql.autoBroadcastJoinThreshold 我对此没有什么问题。

我们暗示作为广播的小数据帧的生命周期是什么?它将在记忆中保留多长时间?我们如何控制它?

例如,如果我使用广播散列连接将一个大数据帧和一个小数据帧连接了两次。当第一个连接执行时,它将向工作节点广播小数据帧,并执行连接,同时避免大数据帧数据的洗牌。

我的问题是,执行器将在多长时间内保留广播数据帧的副本?在会话结束之前,它会保留在内存中吗?否则一旦我们采取行动,它就会被清除。我们能控制或清除它吗?或者我只是在想错误的方向…

2 回复 | 直到 7 年前

1

3

Dave DeCaprio 7 年前

至少在spark 2.4.0中,您的问题的答案是,数据帧将保留在驱动程序进程的内存中,直到sparkContext完成,也就是说,直到应用程序结束。

事实上,广播联接是使用广播变量实现的,但是当使用数据帧API时,您无法访问下属广播变量。Spark本身不会在内部使用后破坏这个变量,所以它只是停留在周围。

具体来说,如果您查看BroadcastExchangeExec的代码( https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala ,可以看到它创建了一个私有变量。 relationFuture 它保存广播变量。这个私有变量只在这个类中使用。作为一个用户,您没有办法访问它来调用destroy,并且curretn实现中没有任何地方会激发您调用destroy。

2

0

luminousmen 7 年前

这里的想法是在连接之前创建广播变量,以便轻松控制它。没有它,你就无法控制这些变量-火花为你做。

例子:

from pyspark.sql.functions import broadcast

sdf2_bd = broadcast(sdf2)
sdf1.join(sdf2_bd, sdf1.id == sdf2_bd.id)

对于所有广播变量(自动在联接中创建或手动创建),应用此规则: