周俊2021-08-16老师， //订单表与用户表关联 select sum(order.price * order.volume), user.id from order inner join user on order.userId = user.id where user.type = ‘Head Users’ group by user.id 这段代码中，catalyst会做谓词下推处理，将过滤条件放到贴近数据源扫描的地方，这里过滤之后user表的大小已经属于小表变成广播变量的阈值了吧，为什么还会有AQE呢作者回复: 好问题~ 推出AQE之前，Spark SQL的静态优化没有那么智能。举例来说，表A：10G；表2：2G，表2过滤之后100MB，然后比如把广播阈值设置为1GB。讲道理，理论上，过滤之后应该选择Broadcast Join才对。 但是，Spark SQL在静态优化期间，会根据表A、B的存储大小，来判断选择什么Join策略，由于表A、B都大于1GB，所以Spark SQL在优化阶段自然选择Shuffle Join。它在优化阶段，并不知道filter之后，表B会变小，小到足以放进广播变量。这个正是静态优化的限制与痛点，是AQE设计初衷之一。 但是AQE就不同，根据Shuffle过后的中间文件，Spark SQL可以动态地决定，把原理的Shuffle Join调整为Broadcast Join。
12
Fendora范东_2021-05-09磊哥，请教下 Join策略调整里面，为啥只有SMJ能降级，而其他Join比如SHJ不能降级作者回复: 好问题~ 其实这个问题我特别想让你独立思考，你先花5分钟想想，结合咱们之前讲过的内容，然后再看答复~ —————————给你留个分割线，哈哈———————————————— 首先，Spark有个参数：spark.sql.join.prefersortmergeJoin，它默认是true，也就是说，对于Shuffle Joins，Spark SQL会prefer也就是偏向优先选择Shuffle Sort Merge Join，换句话说，如果你不动这个参数的话，那么Spark SQL根本不会去考虑用Shuffle Hash Join。这么说吧，在Spark SQL眼里，Shuffle Hash Join的地位非常的低，虽然这一点我不认可，但是Spark就是这么干的。 所以，如果你想启用Shuffle Hash Join，那就必须把spark.sql.join.prefersortmergeJoin置为false。这还不算完，即便如此，Spark SQL其实还是会倾向使用Shuffle Sort Merge Join。当且仅当如下条件成立的时候，Spark才有可能考虑Shuffle Hash Join： 1）外表比内表至少大3倍 2）内表所有数据分片，都要小于广播变量 你说说，这个条件苛刻不？因此，想要启用Shuffle Hash Join，开发者就只剩下一条路可走：Join hints。也就是使用.hint("SHUFFLE_HASH")关键字，来强制Spark SQL选择Shuffle Hash Join。但是这个时候，你都使用强制机制了，Spark SQL会走单独的code path，不用再去利用AQE的Join策略调整、去看看能不能转Broadcast Joins。 因此，总结下来，在AQE的优化机制下，只有SMJ会降级为Broadcast Hash Join，压根跟Shuffle Hash Join没啥关系~
共 3 条评论8
sparkjoy2021-08-27老师，为什么不在join之前统计两个表的大小，从而决定是否用BHJ，而是map结束之后才根据shuffle文件总大小去判断呢？作者回复: spark的计算模式是惰性求值，在没有action的情况，对于数据集大小的判断，其实都是一种基于统计的猜测，这个就是CBO的工作原理。CBO要想work，必须要先用ANALYZE TABLE来统计数据表的各种信息。 AQE相反，放弃了这种思路，不再提取计算统计信息，而是根据运行时的反馈，来动态优化。shuffle本身，你可以把它看作是一种“action”，因为map阶段要落盘，reduce阶段在本质上，是另一个Stage的“map”阶段。下一个stage通过磁盘与上一个stage交换数据。也就是说，Spark能利用的运行时数据，只有map阶段输出的中间文件，对于数据表的大小也好，过滤之后的表大小也好，如果没有shuffle map的中间文件，spark并没有哪里可以获取关于数据表大小、分布这类信息。因此，AQE必须依赖shuffle中间文件，来完成动态的判断。
4
zxk2021-05-07问题一：个人认为 AQE 可以在加载读取文件的时候获取一些运行时信息，或者做 cache 的时候。这里也有个疑问，就是 AQE 会不会根据这些信息也进行一些优化？ 问题二想到两种方法，不知道是否可行： 1. Spark 有一个动态分配调整 executor 的功能，在 Shuffle Map 阶段由 Driver 端汇聚信息决定好倾斜数据的切割方式，之后部分数据发送到原有的 executor 上，切割出来的数据发送到新的 executor 上，同时也需要注意对应做关联的数据也需要复制一份传输到新的 executor 上，但这样会带来 driver 端决策的开销、新的 executor 调度开销以及关联数据额外复制并通过网络传输的开销。 2. 仍按照原来的方式进行，但在 Reduce 阶段切割数据后，起一个新的 executor 来分担切割后的数据，并通知 driver 端。如果能够在同节点上新起 executor，还可以消除网络之间的传输，只做进程间的数据传输即可。 这里想向老师请教一个关于 Join 策略调整的问题，如果 a、b 为事实表，c 为维度表，a、c 做关联后 c 从原来的 SMJ 被 AQE 优化为了 BHJ 后，如果紧接着 b 又跟 c 做关联，那么 Spark 是否会直接使用 BHJ，还是仍需要将 b、c 做 SHuffle Map 之后才能优化为 BHJ？展开作者回复: 先来说说你提的问题，好问题~ 这里面涉及到一个多表Join的执行过程问题，先说结论，无论你的查询涉及多少张表，所有的数据库引擎（包括Spark SQL在内），都是以两表Join为粒度，逐渐地完成Join的。以你的例子来说，a、b、c三张表，a、c先关联，那么这个时候，a、c会生成关联后的中间结果tmp，后续和b表关联的，是这个“tmp表”，已经不再是你说的c表了。因此，回答你的问题，a、c关联利用AQE转化为BHJ，那么后续Spark SQL需要继续关联tmp和b，这个时候，AQE会继续判定tmp结果集大小，如果小于广播阈值，那么它就会把tmp与b的关联，继续转化为BHJ。相反，如果tmp很大，那么Spark SQL就用默认的SMJ。 再回过头来说问题一，现有的AQE机制，是不会利用你说的那些信息的，不过这正是问题所在，就是AQE可不可以、应不应该利用其它的运行时信息，我个人认为，是可以的，只不过这里还是要平衡一个运行时信息统计效率的问题，这个是关键所在~ 关于问题二，你说的特性，应该是Spark的Dynamic allocation，也就是以Executors为粒度，根据任务的计算负载，动态缩放Spark的分布式集群。这个确实是个不错的思路，不过呢，如果仅仅是依赖于集群的缩放，那么其实还是解决不了以task为粒度做负载均衡的问题，因为不论你有多少Executors，倾斜的数据，还是会分发到某一个Executor上面去。这块的答案其实是“两阶段Shuffle”，第一阶段通过加盐， 把数据打散，这样数据就会均衡地分发到集群内所有的Executors，解决了Executors之间的负载均衡问题；第二阶段是去盐化，也就是把第一阶段添加的随机“盐粒”去掉，保证关联关系的一致性。关键两阶段Shuffle的细节，可以参考第29讲哈~
4- sparkjoy2021-08-26老师，cbo的信息是存在表的元信息里吗？
作者回复: 对，CBO相关的元信息，都存储在Hive表，使用ANALYZE语句分析得到的统计信息，都会记录到那些数据表里面去~
2 
To_Drill2021-11-30老师您好，关于文章中的图-两边倾斜中的关联箭头指向有点疑问，如果两张表的关联分区都发生了倾斜，然后都进行了拆分，那么拆分之后的子分区中的是局部数据，应该和对方的全量数据都关联下才能保证原来的关联逻辑，这样的话各个子分区的关联箭头指向是不是应该如下图所示呢？ 拆分-01-拷贝 -> 拆分-01 拆分-01 -> 拆分-02 拆分-02 -> 拆分-01-拷贝 拆分-02-拷贝 -> 拆分-02-拷贝作者回复: 对的，对的~ 老弟说的对~ 是应该像你说的那样，交叉组合，原图画的有问题，回头我改下，感谢老弟提醒~
1
A2021-07-21spark.sql.adaptive.advisoryPartitionSizeInBytes，由开发者指定分区合并后的推荐尺寸。分区合并的最终尺寸这个配置应该是个建议值吧？min(maxTargetSize,advisoryTargetSize)会通过这行代码来取advisoryTargetSize作者回复: 是的，没错，就是你说的这个逻辑~
共 2 条评论1
aof2021-05-141. 如果不用Map阶段的输出文件的话，那应该就是实时统计了吧，比如"ANALYZE TABLE COMPUTE STATISTICS" 2. 为了防止倾斜分区都出现在同一个Executor的情况，可以考虑对倾斜数据的key进行加前缀，然后再将这些数据进行一下重分区repartition()，分区数指定为executor的个数。但是，由于使用了repartition()，也就引入了shuffle开销，这个也是一个要平衡的问题作者回复: 第二题满分💯，没错，要实现Executors级别的均衡，那确实需要加盐来处理，也就是把Join Keys打散，这样哈希过后再分发数据，数据在Executors中的分布就会比较均衡。 第一题实时统计的思路也没问题~ 这个思路其实就是传统DBMS的真正的CBO，实时计算每一步的costs，然后按照costs动态做优化。
1
Stony.修行僧2021-05-07老师请教一个问题，一个10g的csv 文件，里面有4个字段，其中三个字段需要做匿名化。在匿名化三个字段里过程中，可以partition第四个字段来提高性能吗？求老师的意见和建议作者回复: 我先说说对于问题的理解哈，不确定我的理解是不是对的。 你的诉求：一个csv格式的文件，4个字段，目标是对其中3个字段做匿名化，想问按照第四个字段做分布式计算来提高执行效率行不行。 如果是这样的话，那我觉得完全没问题，如果第四个字段分布比较均匀，不存在明显的倾斜问题，我觉得完全可以啊，而且就是应该这么做来提升执行效率。 如果有倾斜的话，你还可以手工加盐处理，手工加盐的方法我们29讲会细说，到时候可以重点关注下~
共 2 条评论1
CRT2022-01-18老师，既然分区合并是要等shuffle map阶段结束之后才可以适用，那么shuffle map阶段原本会生成多少个task，应该怎么决定呢？作者回复: 我们把shuffle之前的stage称作stage0，也就是shuffle map stage，把shuffle过后的stage，成为stage1，也就是reduce阶段的stage。 Stage0的并行度，或者说有多少task，是由stage0里面第一个RDD/DataFrame决定的。而Reduce阶段的stage1，它的并行度，或者说tasks数量，它是由配置项spark.sql.shuffle.partitions决定的