(先区分一个概念,rdd的分区数和rdd的分区器(partitioner)的分区数不是一个)
reduceByKey groupByKey join等,如果没有明确的标记出来分区,那么会调用defaultPartitioner方法生产一个分区器。逻辑如下:
- 如果父rdd有partitioner存在,那么找父rdd的分区数最大的rdd。如果不存在进入2.如果存在进入3.
- 新建一个HashPartitioner,分区个数defaultNumPartitions(如果配置了spark.default.parallelism那么就是该值,否则,父rdd的分区数的最大值)。
3.如果合法或父rdd(第一步选出来的)的partitioner的分区数数大于defaultNumPartitions,那么就用父rdd的partitioner,否则2.
合法的判断:
父rdd(第一步选出来的)的partitioner的分区个数,不小于父rdd(第一步选出来的)分区的1/10。
合法的判断是在2.3加入的,因为在spark有个标准就是尽可能的让并发度大,在上面的判断中可以看出这一点。
本人猜测加入这个的意思可能是为了预防spark.default.parallelism参数配置的太大了,其实有的时候不是说并发度越大越好。加入这个就可以解决这个问题,例如,父rdd的分区最大个数是100,父rdd的partitioner的分区数是20,那就是说当前的并发度是20,20相对于100来说并发度下降了5倍,影响不是很大那么就用父rdd的partitioner。10倍是最大,超过10倍性能下降太多了。