作者回复: Perfect!满分💯,两道题答的都很好~
作者回复: 你是对的,确实会报错。这里我没有交代清楚,RDD确实不能直接用广播变量封装,它不像DataFrame,DataFrame广播的那部分源码在内部把collect这个事做了,所以你可以直接用广播封装DataFrame,但是RDD没有,确实需要先手动collect RDD数据集,然后再在driver端用广播变量封装,我的锅,没有交代清楚~ 不过,歪打正着,通过这个例子,你可以更好地理解Driver在构建广播变量时的计算过程,也就是第一步都是把数据集collect到Driver端,不管是RDD、DataFrame、Dataset,区别无非是collect这件事是谁做的。RDD是开发者来做,而DataFrame、Dataset是Spark自己“偷偷”做了。
作者回复: 非常赞哈👍哈~ 凡是看源码的同学,都先给个赞~ 这块非常值得探讨。我的理解是这样的,代码层面,spark确实已经有code在尝试用p2p的方式来分发广播变量,从而减轻driver负担。 但是,据我观察,这部分代码尚未生效。细节可以参考这个ticket:【Executor side broadcast for broadcast joins】https://issues.apache.org/jira/browse/SPARK-17556,看上去还是进行中的状态。 另外,从代码看,目前还是先用collect拉到driver,然后再分发出去: BroadcastExchangeExec中的relationFuture用于获取广播变量内容 在relationFuture内部: 1. 先是调用executeCollectIterator生成内容relation; 其中,executeCollectIterator调用collect把结果集收集到driver端 2. 然后用sparkContext.broadcast(relation),把生成好的内容广播到各个Executors 并没有看到哪里从Executors拉取数据分片、来减轻driver负载。 并且,这里还有提示driver内存不够的exception: new OutOfMemoryError("Not enough memory to build and broadcast the table to all " + "worker nodes. As a workaround, you can either disable broadcast by setting " + s"${SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key} to -1 or increase the spark " + s"driver memory by setting ${SparkLauncher.DRIVER_MEMORY} to a higher value.").initCause(oe.getCause)) 你贴的这段代码确实在尝试用片2片,不过,需要仔细看看,它在哪里调用,被谁调用~
作者回复: 先来说第一个问题~ a)你说的没错,按理说,应该走广播,毕竟广播阈值默认10MB,你的数据集足够小,磁盘上1MB,内存里4MB,怎么看都比广播阈值小。我能想到的可能的原因,就是Spark SQL对于数据集大小的误判,就是它对于DIM的预估大于10MB。你不妨用下面这个方法,计算一下DIM在Spark SQL下的预判大小: val df: DataFrame = _ df.cache.count val plan = df.queryExecution.logical val estimated: BigInt = spark .sessionState .executePlan(plan) .optimizedPlan .stats .sizeInBytes 看看能不能发现什么端倪。虽然咱们不能完全确认原因,不过要解决这个问题倒是蛮简单的。一来可以用broadcast函数,二来可以用join hints,总之就是各种花式强制广播。 b)这里就需要更多信息来判断了,6G是从哪里得到的?Spark UI的DAG图吗?sort阶段的6G,仅仅是DIM表的大小?可以加我微信“方块K”或是“rJunior”,把完整的DAG贴给我~
作者回复: 思考的很深入,👍赞一个~ 1. 我理解,你问的是这句吧?“由于左右表的分区数是一致的,因此 Shuffle 过后,一定能够保证 userID 相同的交易记录和用户数据坐落在同一个 Executors 内。” HadoopRDD的分区数、或者说并行度,确实是由HDFS文件系统决定的;但是,Shuffle过后,每个分布式数据集的并行度,就由参数spark.sql.shuffle.partitions来决定了,这个咱们在配置项哪一讲说过哟~ 因此,如果你没有手工用repartition或是Coalesce去调整并行度,默认情况下,大家Shuffle过后(在Reduce阶段)都是这个并行度。 2. 默认确实是开启的,默认值确实也是10MB,但是,这个10MB太太太太太太(太 x N)小了!很多小表其实都超过了这个阈值,因此,如果你懒得去调整这个参数,可以直接用broadcast(userDF)这种强制广播的方式,省时省力,比较方便~
作者回复: 这里其实有两个容易混淆的概念哈~ 一个是并行度,并行度其实是从数据角度出发,表示的是你的分布式数据集划分的粒度,再直白点说,它和分区数是等效的。因此,它其实跟你集群有多少Executors,每个Executors配置了多少cores,没有关系~ 第二个是并发度,或者叫Executors线程池大小,也就是你用spark.executor.cores类似的参数,给Executors指定的cores资源。它限制了在同一时间,你的Executors中最多同时能跑多少个任务。Executors并发度乘以集群中的Executors数量,其实就是你集群的并发处理能力,很多地方也叫并行处理能力。其实蛋疼的地方在于,不同的作者、不同的上下文,并发和并行这两个词,总是混用。所以也就造成大家都比较困惑。 咱们在配置项第一讲,其实就在尝试厘清、约定这两个词的定义,一来方便大家理解,二来方便后续讨论。 所以,回答你的问题,其实没什么不健康的哈~ 10000并行度,意味着10000个分区的分布式数据集,这个应该不难见到。另外100个cores的集群,其实也不算小了~ 不过你说的2000任务我没有get到,不知道是2000并行度,还是2000的集群并发。如果是2000集群并发的话,这个数和100cores对不上。这意味着你的每个core需要20个超线程,哈哈,目前还没有这么给力的CPU。一般CPU也就2个超线程。
作者回复: 能否广播,取决于b表的存储大小,是否小于广播阈值,也就是:spark.sql.autoBroadcastJoinThreshold。如果小于这个阈值,就会广播,否则就不会。 如果懒得设阈值,还可以利用 API 强制广播,这里的具体细节,可以参考第13讲哈,就是后面的一讲~ 会详细说,怎么把Shuffle Join,转化为Broadcast Join
作者回复: 是的,以小博大
作者回复: 可以参考Bennan同学的答案哈: 1. P2P思路:改成由driver获取到数据分布,然后通知各个executor之间进行拉取,这样可以利用多个executor网络,避免只有driver组装以后再一个一个发送效率过低 2.当两个需要join的数据集都很大时,使用broadcast join需要将一个很大的数据集进行网络分发多次,已经远超出了shuffle join需要传输的数据
作者回复: 关于Cache(Persist)的部分,建议老弟关注第16讲:内存视角(二):如何有效避免Cache滥用?这一讲,比较系统、细致地介绍了Cache的使用场景、原则,和一般注意事项,尤其是什么时候该Cache,什么时候不能滥用Cache,老弟可以先看看哈~