RDD分区的设定规则涉及多个因素,具体如下:

(1)local模式
默认并行度取决于本地机器的核数,即
local: 没有指定CPU核数,则所有计算都运行在一个线程当中,没有任何并行计算
local[K]:指定使用K个Core来运行计算,比如local[2]就是运行2个Core来执行
local[*]: 自动帮你按照CPU的核数来设置线程数。比如CPU有4核,Spark帮你自动设置4个线程计算

(2)集群模式
集群模式包含Stanalone、Yarn模式,Mesos的默认并行度为8
默认并行度取决于所有executor上的总核数与2的最大值,比如集群模式的设置如下:
--num-executors 5
--executor-cores 2
上面配置Executor的数量为5,每个Executor的CPU Core数量为2,
executor上的总核数10,则默认并行度为Max(10,2)=10。
 

一、默认分区规则

  1. 从本地文件创建
    分区数默认为: $$ \text{min}( \text{defaultParallelism}, 2 ) $$ 其中 defaultParallelism 由集群配置决定(如 spark.default.parallelism)。

  2. 从HDFS文件创建
    分区数等于HDFS文件的块数(Block),每个块对应一个分区。

  3. 通过parallelize()创建
    分区数由 spark.default.parallelism 参数控制。若未配置:

    • 本地模式:等于CPU核心数
    • 集群模式:等于所有Executor的总核心数

二、分区调整方法

  1. 手动指定分区数
    在创建RDD时通过参数直接指定:

    sc.textFile("path", numPartitions = N)
    sc.parallelize(data, numSlices = N)
    

  2. 转换操作后的分区变化

    • filter()map()等操作:继承父RDD的分区数
    • union():分区数为父RDD分区数之和
    • join()groupByKey():默认使用哈希分区,分区数由 spark.default.parallelism 决定
  3. Shuffle操作的分区控制
    通过参数指定Shuffle后的分区数:

    rdd.reduceByKey(func, numPartitions = N)
    rdd.repartition(N)  // 强制重分区
    rdd.coalesce(N)      // 合并分区(无Shuffle)
    


三、分区数优化原则

  1. 分区数与并行度
    每个分区对应一个Task,分区数应满足: $$ \text{总分区数} \geq \text{集群总核心数} \times 2 $$ 避免核心闲置或任务排队。

  2. 分区大小建议
    单个分区数据量建议在 128MB~1GB 之间。过大易导致OOM,过小增加调度开销。

  3. 动态调整场景

    • 数据倾斜时:对Key重新分区(如 repartition
    • 迭代计算时:通过 persist() 缓存合适的分区数

四、关键配置参数

参数名 默认值 作用
spark.default.parallelism CPU核心数 控制RDD默认分区数
spark.sql.shuffle.partitions 200 控制DataFrame的Shuffle分区数

提示:实际分区数需结合数据规模、集群资源和计算逻辑动态调整。可通过 rdd.getNumPartitions 查看当前分区数。

Logo

开源鸿蒙跨平台开发社区汇聚开发者与厂商,共建“一次开发,多端部署”的开源生态,致力于降低跨端开发门槛,推动万物智联创新。

更多推荐