13 | 广播变量（二）：如何让Spark SQL选择Broadcast Joins？

作者回复: Perfect x 2！两道题都是满分💯~ 不过，第一题，我再追问一句，当然，这么追问有点过分，哈哈，毕竟咱们这节课还没有讲不同Join的实现方式（26讲会展开）。追问的问题是： SHUFFLE_MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL 这几个hints的作用和效果，分别是什么？他们分别适合哪些场景？不妨提前思考一下，等到了26讲，咱们再细聊哈~

作者回复: 好问题~ 这里我先说声抱歉，看到你这个问题，我意识到我原文没有说清楚。是这样的，我们分两种情况来说。为了叙述方便，我把你的数据集记为D，也就是磁盘中133.4MB，展开到内存是1000MB。 第一种情况，数据集D上不存在Cache，这个时候那D直接参与后续的计算，比如Join，由于Spark SQL无从判断D在内存中的大小，因此它会“偷懒”地直接拿磁盘大小作为判断依据，也就是你上面说的那种情况，广播阈值大于133.4MB，就可以广播，反之则不行。不得不说，这个“懒”偷的有点过分，不过现阶段Spark SQL就是这么做的，这个和Spark的版本没有关系，3.0中Spark SQL也是这么处理的。 第二种情况，数据集之上有Cache，这个时候，Spark SQL判定的依据，就是Cache中的存储大小，也就是1000MB，所以你不妨再做个实验，也就是用Cache过后的D再去做Join，这个时候，你会发现，广播阈值必须要大于1000MB才行。 总结下来，对于数据集大小的判定，Cache与否完全不同，不Cache则直接用磁盘存储大小，Cache过后则使用Cache中的存储大小。需要特别注意的是，第一种情况，也就是不加Cache的时候，Spark SQL简单粗暴地直接用磁盘存储大小，坦白讲是存在隐患的，也就是创建广播变量创建过程中Driver端的OOM问题，或是广播变量创建过程中的Time out问题。因此，即便在那种情况之下，Spark SQL会选择参考磁盘存储大小，但是我的建议还是事先估算好数据集在内存中的存储大小，做到有备无患。

作者回复: 好问题， 这种写法，Spark不会复用先前的广播变量，所以第二次的Broadcast会重复计算。 复用广播最保险的方式，是这种写法： val bcSmallTable = sparkContext.broadcast(smallTable) bigTableA.Join(bcSmallTable.value, ...); bigTableB.Join(bcSmallTable.value, ...);

作者回复: 好问题~ 这主要是因为left join的计算逻辑。简单的回答是，广播左表、或者说基表，没用！ 为什么这么说呢，你不妨仔细想想left join的原理。它是把inner join的结果，再加上不满足关联条件的结果。 如果是广播右表的话，左表的数据分区，能看到右表的全量数据，不管是满足关联条件和不满足关联条件，左表的数据分区都能立即得到答案。 但是，反过来不行，比如把左表广播了，右表“待在原地、保持不动”，那对于左表来说，对于每一个右表的数据分区来说，左表没有全局视角，它只能知道哪些右表分区的数据满足关联条件，但是，它不知道在全局情况下，到底有没有不满足关联条件、需要生成null的数据记录。 可能说的有点绕，你不妨仔细想想left join的计算原理，就能明白，在left join下，广播左表是没有用的；同理，right join，广播右表，也没有用。因为他们都不能实现left/right join原本的计算逻辑~

作者回复: 是这样的，每个广播的上限是8G，超过这个限制，Spark直接抛异常。所以，这个8G是以广播变量为粒度的。15个小表的话，只要他们各自不超过8G就没事。不过，话说，同时搞15个广播变量，Driver不一定受得了~ 当然，要看你的每个广播变量有多大了。 不过anyway，只要你Driver内存足够大，每个广播又不超过8G，就没事。

作者回复: 具体细节可以参考这里哈：https://databricks.com/blog/2015/04/28/project-tungsten-bringing-spark-closer-to-bare-metal.html java.lang.String object internals: OFFSET SIZE TYPE DESCRIPTION VALUE 0 4 (object header) ... 4 4 (object header) ... 8 4 (object header) ... 12 4 char[] String.value [] 16 4 int String.hash 0 20 4 int String.hash32 0 Instance size: 24 bytes (reported by Instrumentation API)

作者回复: 最早在哪里看到的，说实话我也忘了[允悲]，太久了。不过这个方法一定是切实可行的，日常的工作中验证无数次了哈~ 需要提醒的是，要预估数据集的内存占用，需要先把数据集cache起来，否则的话，sizeInBytes给出的，实际上源文件在磁盘上的存储大小，切记切记

作者回复: 不会的~ 和语法没什么关系，主要是看表大小。

作者回复: 能把执行计划贴出来吗？（可以在join完成之后的DataFrame之上调用explain） 我想知道这4张表的关联顺序，讲道理，应该是: ((a inner join b) inner join c) left join d 这是我的理解，如果是这样的话，理论上应该都是Broadcast Join才对，可以把explain执行计划贴出来确认一下~

作者回复: 这个配置项spark.sql.autoBroadcastJoinThreshold，它的含义是，size不大于这个设置的数据集，可以被广播。从log来看，你的广播变量小于1G，所以没有问题~ 并不是说你设置多少，广播变量大小就是多少