为什么 Spark 广播交换数据大小大于加入时的原始大小？

Question

我正在对两个表 A 和 B 进行广播连接。 B 是使用以下 Spark SQL 创建的缓存表：

create table B as select segment_ids_hash from  stb_ranker.c3po_segments
      where
        from_unixtime(unix_timestamp(string(dayid), 'yyyyMMdd')) >= CAST('2019-07-31 00:00:00.000000000' AS TIMESTAMP)
      and
        segmentid_check('(6|8|10|12|14|371|372|373|374|375|376|582|583|585|586|587|589|591|592|594|596|597|599|601|602|604|606|607|609|610|611|613|615|616)', seg_ids) = true
cache table B

'segment_ids_hash' 列是整数类型，结果包含 3640 万条记录。缓存的表大小约140MB，如下图

然后我按如下方式进行了连接：

select count(*) from A broadcast join B on A.segment_ids_hash = B.segment_ids_hash

这里广播交换数据大小约为 3.2 GB。

我的问题是为什么广播交换数据大小 (3.2GB) 比原始数据大小 (~140 MB) 大得多。什么是间接费用？有没有办法减少广播交换数据大小？

谢谢

Answer 1

; 博士：我也在学习数据大小指标的来源。这可能只是估计的操作大小，它可能无法反映数据的实际大小。暂时不要太担心。

完整版本：

更新：回来纠正一些错误。我看到以前的答案缺乏一些深度，所以我会尽量深入挖掘这个问题（我对回答问题仍然相对较新）。

更新 2：重新措辞，删除一些过分的笑话（sry）

好的，所以这件事可能很长，但我认为这个指标并不是数据的直接大小。

首先，我对此进行了测试运行，以使用 200 个执行程序和 4 个内核重现结果：

这返回了这个结果：

现在我发现有些东西很有趣，因为我测试的 dataSize 大约是 1.2GB 而不是 3.2GB，这让我阅读了 Spark 的源代码。

当我去github时，我看到BroadcastExchange中的4个数字对应于这个：第一个链接：BroadcastHashJoinExec：https : //github.com/apache/spark/blob/master/sql/core/src/main/scala/org /apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala

这一项对应的数据大小：

我发现这里的关系 val 似乎是一个 HashedRelationBroadcastMode。

转到 HashedRelation https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala：

因为我们有 Some(Numrows) （它是 DF 的行数）。比赛案例用例一（第 926:927 行）

回到 HashedRelation 构造函数-y 部分：

由于连接用于散列整数，因此类型不是 Long => 连接使用 UnsafeHashedRelation

到 UnsafeHashedRelation：

现在我们转到 UnsafeHashedRelation 中确定估计大小的地方，我发现了这个：

关注估计的大小，我们的目标是binaryMap对象（后面的代码assign map = binaryMap）

然后它转到这里：

binaryMap 是一个 BytestoBytesMap，对应这里https://github.com/apache/spark/blob/master/core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java

跳转到 getTotalMemoryConsumption 方法（获取估计大小的方法），我们得到：

这是我目前的死胡同。只是我的两美分，我不认为这是一个错误，而只是连接的估计大小，而且由于这是一个估计大小，我真的不认为它必须非常准确（是的，但是很奇怪在这种情况下诚实，因为差异非常大）。

如果您想继续使用这个数据大小。一种方法是通过修改其构造函数的输入来直接影响 binaryMap 对象。回头看看这个：

有两个变量可以配置，分别是 MEMORY_OFFHEAP_ENABLED 和 BUFFER_PAGE 大小。也许您可以在 spark-submit 期间尝试使用这两种配置。这也是为什么即使您更改了执行程序和内核数量，BroadcastExec 大小也不会改变的原因。

所以总而言之，我认为数据大小是由某种迷人机制生成的估计（这个我也在等待有更多专业知识的人在我深入研究时对此进行解释），而不是您在中提到的直接大小第一张图片 (140 MB)。因此，花费太多时间来减少此特定指标的开销可能不值得。

一些奖金相关的东西：

https://jaceklaskowski.gitbooks.io/mastering-spark-sql/spark-sql-SparkPlan-BroadcastExchangeExec.html

https://jaceklaskowski.gitbooks.io/mastering-spark-sql/spark-sql-UnsafeRow.html