在 PySpark 中，spark.default.parallelism 是一个关键参数，直接影响作业的并行度和资源利用率。
通过合理设置 spark.default.parallelism 并结合数据特征调整，可显著提升 PySpark 作业的并行效率和资源利用率。建议在开发和生产环境中进行多轮基准测试以确定最优值。以下是如何通过调整此参数优化性能的详细说明，结合案例和最佳实践：

---

1. 参数作用与问题场景

参数意义

• spark.default.parallelism 决定了：
  • 新生成的 RDD 的默认分区数（如 sc.parallelize()）。
  • Shuffle 操作（如 groupByKey、reduceByKey）后的分区数（如果未显式指定）。
• 默认值：
  • 本地模式：CPU 核心数（local[*] 时为逻辑核心数）。
  • 集群模式：max(2, total_executor_cores)（YARN/Mesos/K8s）。

常见性能问题

1. 分区过少：
   • 数据倾斜（少数分区处理大量数据）。
   • 无法充分利用集群资源（并行度低）。
   • 频繁的磁盘溢出（单个分区数据过大）。
2. 分区过多：
   • 调度开销增大（大量小任务）。
   • 增加 Shuffle 网络传输压力。
   • 潜在的内存溢出（如广播变量重复分发）。

---

2. 优化策略与设置方法

合理值计算

• 经验公式：

  python">spark.default.parallelism = max(# 基础值：总核心数 × 2~4 倍（IO密集型取高值，CPU密集型取低值）total_executor_cores * 3,  # 确保至少与数据输入分区数对齐（如 HDFS 文件块数）input_partitions  
  )
  

  • 例如：集群总资源为 100 个核心 → 建议设置为 300（3 倍核心数）。

• 动态调整：

  • 如果数据量极大（如 TB 级），可进一步提高至 total_executor_cores * 4。
  • 使用 df.rdd.getNumPartitions() 检查输入数据的分区数。

设置方式

• 代码中配置：

  python">from pyspark.sql import SparkSession
  spark = SparkSession.builder \.config("spark.default.parallelism", "300") \.getOrCreate()
  

• 提交作业时指定：

  spark-submit --conf spark.default.parallelism=300 app.py
  

---

3. 案例分析与验证

场景描述

• 问题作业：处理 1TB 的日志数据，进行 groupByKey 后聚合。
• 原始配置：spark.default.parallelism=200（集群总核心数 100）。
• 症状：
  • Shuffle 阶段耗时 2 小时，某些 Task 处理时间超过 30 分钟。
  • Executor 的 CPU 利用率低于 40%。

优化步骤

1. 参数调整：

   spark-submit --conf spark.default.parallelism=300 --conf spark.sql.shuffle.partitions=300 app.py
   

2. 代码优化：
   • 将 groupByKey 替换为 reduceByKey（减少 Shuffle 数据量）。
   • 添加 Salt 处理倾斜键（如 key -> f"{key}_{random.randint(0,9)}"）。

结果对比

指标	优化前	优化后
总执行时间	4.2 小时	1.8 小时
最长 Task 耗时	32 分钟	8 分钟
Executor CPU 利用率	35%	75%

---

4. 注意事项与进阶技巧

注意事项

• 与 spark.sql.shuffle.partitions 联动：
  • 针对 DataFrame/SQL 操作，需同时设置此参数（默认 200）。

  python">spark.conf.set("spark.sql.shuffle.partitions", "300")
  

• 避免过度分区：
  • 检查任务执行计划（df.explain()），确保没有生成过多小文件。
  • 监控 Spark UI 中的 Task 耗时分布（避免 99% 的任务在 1 秒内完成）。

进阶优化

1. 动态分区控制：
   • 对倾斜数据使用 repartition(n) 或 coalesce() 显式调整。
   • 使用 spark.adaptive.enabled=true（Spark 3.0+ 自适应查询优化）。
2. 数据本地性：
   • 确保输入数据分区与 HDFS 块分布对齐（避免跨节点传输）。
3. 硬件资源匹配：
   • 每个分区的数据量建议在 128MB~1GB 之间（HDFS 块大小对齐）。

---

5. 监控与调优验证

1. 监控工具：
   • Spark UI（任务时间分布、Shuffle 读写量）。
   • Ganglia/Prometheus（集群级 CPU/内存/网络监控）。
2. 调优验证：
   • 逐步增加 spark.default.parallelism，观察任务时间变化曲线。
   • 使用 spark.dynamicAllocation.enabled=true 自动扩展 Executor。