Spark文件读取任务数与分区数、shuffle配置不一致的原因咨询

问题背景 #

已知任务数与分区数相关：

• 非文件读取类任务（如Join的Merge阶段），任务数和输入表分区数一致，且遵循spark.conf.set('spark.sql.shuffle.partitions',X)配置。
• 但文件读取任务（如Parquet扫描）不匹配：比如一张由258个Parquet文件组成的表，全量读取仅使用8个任务，且和shuffle配置不符（假设X≠8）。

执行以下代码：

df=spark.sql('select * from transaction')
df.rdd.getNumPartitions()

返回分区数为57，说明单个Parquet文件不等于一个分区。但后续用该表做Join并分组操作时，仅使用8个任务。

疑问：为何Spark在文件数258、分区数57的情况下，选择使用8个任务，且不遵循spark.sql.shuffle.partitions配置？

原因解析 #

1. 文件读取阶段的任务数逻辑 #

Spark读取Parquet这类列式存储文件时，不是按文件数直接对应分区，而是基于数据块大小和文件合并规则切分：

• 依赖spark.sql.files.maxPartitionBytes（默认128MB）控制单个分区的最大字节数，spark.sql.files.openCostInBytes（默认4MB，衡量打开文件的开销）决定小文件是否合并。你这里258个文件生成57个分区，就是因为大量小文件被合并成了单个分区。
• 你看到的全量读取仅用8个任务，核心是集群资源并行度限制：如果集群Executor总核数仅为8，Spark实际执行的并行任务数不会超过可用CPU核数，即便分区数是57，也会分批执行，表现为同时运行8个任务。

2. Join分组阶段用8个任务的原因 #

spark.sql.shuffle.partitions仅控制shuffle后的默认分区数，但以下场景会覆盖这个配置：

• 如果是广播Join：小表被广播到各Executor，大表侧无需触发shuffle，任务数由大表读取分区数决定，再受集群并行度限制，最终用8个任务执行。
• 自适应执行（Adaptive Execution）默认开启：Spark会根据实际数据量、资源情况动态调整分区数。如果shuffle后数据量极小，会自动合并分区，把默认的shuffle分区数（通常200）合并为更少的数量（比如8个），直接覆盖spark.sql.shuffle.partitions的配置。

3. 关键参数总结 #

• 控制文件读取分区数：spark.sql.files.maxPartitionBytes、spark.sql.files.openCostInBytes
• 控制shuffle分区数：spark.sql.shuffle.partitions（会被自适应执行覆盖）
• 限制并行任务数：spark.executor.cores、spark.executor.instances

内容的提问来源于stack exchange，提问作者Eugenio.Gastelum96