Spark 参数调优实战：从 spark-submit 到内存与并行度配置

1. 为什么 spark-submit 的参数总让人一头雾水刚接触 Spark 那会儿我最怕的就是写 spark-submit 命令。倒不是命令本身有多复杂而是那一长串参数——--executor-memory、--num-executors、--driver-memory、--executor-cores、spark.default.parallelism、spark.sql.shuffle.partitions——每一个看起来都认识但组合到一起就不知道从哪下手了。更让人抓狂的是同一个任务参数调对了跑得飞快调错了要么 OOM 直接挂掉要么资源利用率低得可怜集群上一大半节点在摸鱼。这篇文章就是把我这些年踩过的坑、调过的参数、总结出来的经验一次性讲清楚。不管你是刚上手 Spark 的新手还是已经写过一些作业但总感觉性能差口气的开发者都能从这里找到可以直接抄作业的配置方案。我不会只告诉你“把这个参数设大一点”而是会解释清楚每个参数背后的逻辑——它控制什么、影响什么、什么场景下该调、调到多少合适。核心关键词就一个spark 参数。围绕它我们把 Driver、Executor、并行度、内存模型、Shuffle 这几大块全部拆开揉碎。先打个预防针Spark 的参数体系分为三个层次——集群配置参数spark-env.sh 或集群管理器层面、提交参数spark-submit 命令行、代码内参数SparkConf 或 SparkSession.builder。三者的优先级是代码内 提交命令行 配置文件。很多人调参调了半天没生效就是因为没搞清楚这个优先级关系。后面我会在具体参数里反复强调这一点。2. 核心参数全景图从提交命令到运行时配置2.1 spark-submit 命令行参数速查先上一张表把最常用的提交参数列出来后面再逐个展开。参数作用常用值优先级--master指定集群管理器地址yarn、local[*]、spark://host:7077命令行--deploy-modeDriver 运行位置client、cluster命令行--driver-memoryDriver 内存2g、4g、8g命令行--executor-memory每个 Executor 内存4g、8g、16g命令行--executor-cores每个 Executor 的 CPU 核数2、4、5命令行--num-executorsExecutor 数量根据集群规模算命令行--conf任意 Spark 配置项keyvalue命令行--jars额外依赖 jar 包逗号分隔路径命令行--files分发到各节点的文件逗号分隔路径命令行这张表看着简单但每一个参数背后都有一堆门道。比如--executor-cores设多少直接决定了你一个 Executor 能并行跑几个 Task进而影响内存分配策略。再比如--num-executors不是你想设多少就设多少得看集群总资源和队列限制。2.2 资源类参数Driver 与 Executor 的分配逻辑Driver 是 Spark 作业的“大脑”负责解析代码、生成 DAG、调度 Task、收集结果。--driver-memory默认是 1g但在实际生产中这个值往往不够。什么时候需要调大 Driver 内存几个典型场景作业里用了collect()把大量数据拉回 Driver、用了广播变量broadcast且变量很大、或者 DAG 特别复杂导致 Driver 端元数据膨胀。我一般起步设 2g如果发现 Driver 端频繁 Full GC 或者直接 OOM就往上加到 4g 甚至 8g。Executor 是真正干活的“工人”。--executor-memory控制每个 Executor 的堆内存--executor-cores控制它能同时跑几个 Task。这两个参数是耦合的——同样的内存核数越多每个核分到的内存越少容易 OOM核数越少并行度不够资源浪费。经验公式是这样的每个 Executor 的核数建议 2~5 个最常用的是 4。低于 2 浪费内存高于 5 会导致 HDFS 并发写入压力过大每个核可能开一个 HDFS 连接。--num-executors的计算更讲究。假设集群有 10 个节点每个节点 32 核 128G 内存YARN 可用资源大概是总资源的 80%要留一部分给系统那就是 8 节点 × 25 核 × 100G 左右。如果每个 Executor 用 4 核 8G那每个节点能起 6 个 Executor25/4≈6总共 48 个 Executor。但实际还要看队列的maxContainer限制和yarn.nodemanager.resource.memory-mb配置。这里有个很多人忽略的点Executor 内存包含堆外内存off-heap和堆内存on-heap。--executor-memory设的是堆内存但 Spark 还会用一部分堆外内存做序列化、网络传输等。所以实际占用的容器内存 堆内存 spark.executor.memoryOverhead默认是堆内存的 10%最小 384M。如果你设--executor-memory 8g那容器实际申请的是 8g max(384m, 800m) ≈ 8.8g。这个细节在 YARN 上特别重要因为 YARN 是按容器总内存来分配资源的算错了就会导致 Executor 起不来。2.3 并行度参数决定作业快慢的关键并行度是 Spark 性能调优里最核心的概念之一。简单说并行度就是同时有多少个 Task 在跑。并行度太低集群资源用不满任务跑得慢并行度太高Task 调度开销大Shuffle 数据碎片化反而更慢。Spark 里有两个关键的并行度参数spark.default.parallelism控制 RDD 的默认分区数。如果不设对于reduceByKey、groupByKey等操作默认取所有 Executor 的总核数对于parallelize默认取本地核数。spark.sql.shuffle.partitions控制 Spark SQL 中 Shuffle 操作join、group by 等的默认分区数默认值是 200。这个默认值坑了无数人——数据量小的时候 200 个分区导致大量小文件和小 Task数据量大的时候 200 个分区又导致每个 Task 处理太多数据。我一般这样设spark.default.parallelism设为 Executor 总核数的 2~3 倍spark.sql.shuffle.partitions根据数据量来定经验值是每 100MB~200MB 数据一个分区。比如 Shuffle 数据量 50GB那就设 250~500 个分区。如果拿不准可以先设一个值跑一遍看 Spark UI 里每个 Task 的处理时间和数据量再调整。注意spark.sql.shuffle.partitions只对 Spark SQL 的 Shuffle 生效对 RDD 的reduceByKey等不生效。RDD 的 Shuffle 分区数由spark.default.parallelism或算子里的numPartitions参数控制。2.4 内存与序列化参数避免 OOM 的第一道防线Spark 的内存模型经历过几次大改现在2.x 以后用的是统一内存管理模型。每个 Executor 的堆内存分为三块Reserved Memory固定 300MB给 Spark 内部对象用不参与管理。User Memory默认占剩余内存的 40%给用户自定义数据结构、UDF 等用。Spark Memory默认占剩余内存的 60%再分为 Storage Memory 和 Execution Memory各占 50%。相关参数参数默认值说明spark.memory.fraction0.6Spark Memory 占总内存减去 300MB的比例spark.memory.storageFraction0.5Storage Memory 占 Spark Memory 的比例spark.serializerJavaSerializer序列化器建议改成 KryoSerializerspark.kryoserializer.buffer.max64mKryo 序列化缓冲区最大值spark.serializer这个参数我强烈建议每个作业都设成org.apache.spark.serializer.KryoSerializer。Java 原生序列化又慢又占空间Kryo 通常能省 2~5 倍内存速度快好几倍。唯一需要注意的是Kryo 要求注册自定义类才能达到最佳效果不过即使不注册它也能正常工作只是会多存一些类名信息。val conf new SparkConf() .set(spark.serializer, org.apache.spark.serializer.KryoSerializer) .set(spark.kryoserializer.buffer.max, 128m) .registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))2.5 Shuffle 参数性能杀手还是加速器Shuffle 是 Spark 里最耗资源的操作没有之一。它涉及磁盘 I/O、网络传输、序列化/反序列化任何一个环节出问题都会拖慢整个作业。跟 Shuffle 相关的参数很多挑几个最关键的spark.shuffle.file.bufferShuffle write 阶段的缓冲区大小默认 32KB。适当调大到 64KB 或 128KB 可以减少磁盘 I/O 次数。spark.reducer.maxSizeInFlightShuffle read 阶段每次拉取的数据量默认 48MB。调大到 96MB 可以减少网络往返次数。spark.shuffle.compress是否压缩 Shuffle 输出默认 true。保持开启。spark.shuffle.spill.compressSpill 到磁盘时是否压缩默认 true。保持开启。spark.shuffle.io.maxRetriesShuffle 拉取失败重试次数默认 3。网络不稳定时可以调到 10。spark.shuffle.io.retryWait重试等待时间默认 5s。可以配合上一个参数调整。还有一个容易被忽略的参数spark.shuffle.manager。Spark 1.2 以后默认是sort还有一种tungsten-sort基于 UnsafeShuffleManager在特定场景下性能更好。不过 2.x 以后 tungsten-sort 已经合并到 sort 里了不需要单独设。3. 不同场景下的参数配置实战3.1 小数据量作业别让 200 个分区拖后腿我见过太多人写一个处理几 MB 数据的 Spark 作业跑起来要好几分钟。一看 Spark UI200 个 Task每个 Task 处理几十 KB 数据光调度开销就占了大半时间。这就是spark.sql.shuffle.partitions默认值 200 惹的祸。小数据量作业Shuffle 数据 1GB的配置思路是减少分区数降低调度开销。spark-submit \ --master yarn \ --deploy-mode cluster \ --driver-memory 2g \ --executor-memory 4g \ --executor-cores 2 \ --num-executors 4 \ --conf spark.sql.shuffle.partitions10 \ --conf spark.default.parallelism8 \ --conf spark.serializerorg.apache.spark.serializer.KryoSerializer \ my-job.jar这里spark.sql.shuffle.partitions10是根据数据量估算的——假设 Shuffle 数据 500MB每个分区 50MB 左右10 个分区刚好。spark.default.parallelism8对应 4 个 Executor × 2 核 8 核取 1 倍即可。实操心得小作业不要开太多 Executor4 个足够了。Executor 多了反而增加 Driver 的调度负担而且 YARN 分配容器本身也要时间。3.2 中等规模 ETL平衡并行度与资源利用率中等规模 ETLShuffle 数据 10GB~100GB是最常见的场景。这时候参数配置的核心是让每个 Task 处理的数据量在一个合理区间——通常 100MB~200MB 比较合适。假设 Shuffle 数据量 50GB按每分区 128MB 算需要约 400 个分区。Executor 配置方面假设集群有 6 个节点每节点 16 核 64GYARN 可用 12 核 48G。每个 Executor 用 4 核 8G那每节点能起 3 个 Executor总共 18 个 Executor总核数 72。spark.default.parallelism设为 72 × 2 144。spark-submit \ --master yarn \ --deploy-mode cluster \ --driver-memory 4g \ --executor-memory 8g \ --executor-cores 4 \ --num-executors 18 \ --conf spark.sql.shuffle.partitions400 \ --conf spark.default.parallelism144 \ --conf spark.serializerorg.apache.spark.serializer.KryoSerializer \ --conf spark.shuffle.file.buffer64k \ --conf spark.reducer.maxSizeInFlight96m \ --conf spark.shuffle.io.maxRetries5 \ my-etl-job.jar这套配置我用了很多次在中等规模 ETL 场景下基本不需要大改。唯一需要微调的是spark.sql.shuffle.partitions根据实际 Shuffle 数据量动态调整。3.3 大规模 Join 与聚合内存和分区的双重考验大规模 Join比如两张亿级表关联是最容易出问题的场景。常见症状是 Executor OOM、Shuffle 拉取失败、任务跑几个小时甚至跑不完。这种场景下参数配置要围绕三个目标增大内存、增加分区、减少单 Task 压力。--executor-memory直接上 16g 或 32gspark.memory.fraction可以适当调高到 0.7~0.8给 Spark Memory 更多空间。spark.sql.shuffle.partitions按每分区 64MB~128MB 来算宁可多一些也不要少。分区多了最多是调度开销大一点分区少了直接 OOM。开启spark.sql.adaptive.enabledtrueSpark 3.0让 Spark 自动做自适应查询优化包括自动合并小分区、处理数据倾斜。如果确定有数据倾斜可以开spark.sql.adaptive.skewJoin.enabledtrue。spark-submit \ --master yarn \ --deploy-mode cluster \ --driver-memory 8g \ --executor-memory 16g \ --executor-cores 4 \ --num-executors 40 \ --conf spark.sql.shuffle.partitions800 \ --conf spark.default.parallelism320 \ --conf spark.memory.fraction0.7 \ --conf spark.sql.adaptive.enabledtrue \ --conf spark.sql.adaptive.skewJoin.enabledtrue \ --conf spark.serializerorg.apache.spark.serializer.KryoSerializer \ --conf spark.shuffle.io.maxRetries10 \ --conf spark.shuffle.io.retryWait10s \ big-join-job.jar注意spark.sql.adaptive.enabled在 Spark 3.0 默认是 false3.2 以后默认是 true。如果你用的是 3.0~3.1需要手动开启。开启后spark.sql.shuffle.partitions的初始值仍然重要因为自适应优化是在初始分区基础上做合并初始值太小就没法拆分了。3.4 流处理作业稳定压倒一切Spark Streaming 或 Structured Streaming 作业的参数配置逻辑跟批处理不太一样。流处理的核心诉求是稳定——每个批次按时处理完不要积压不要 OOM。关键参数spark.streaming.backpressure.enabledtrue开启背压机制根据上一批次的处理速度自动调整下一批次的接收速率。spark.streaming.kafka.maxRatePerPartition每个 Kafka 分区每秒最大消费条数配合背压使用。spark.streaming.stopGracefullyOnShutdowntrue优雅停止确保当前批次处理完再退出。spark.streaming.blockIntervalReceiver 模式下每个 Block 的生成间隔默认 200ms。调大到 500ms~1s 可以减少 Task 数量。流处理作业的 Executor 数量一般不需要太多4~8 个足够关键是每个 Executor 的内存要够因为流处理是长期运行的内存泄漏和 GC 问题会被放大。spark-submit \ --master yarn \ --deploy-mode cluster \ --driver-memory 4g \ --executor-memory 8g \ --executor-cores 2 \ --num-executors 6 \ --conf spark.streaming.backpressure.enabledtrue \ --conf spark.streaming.kafka.maxRatePerPartition1000 \ --conf spark.streaming.stopGracefullyOnShutdowntrue \ --conf spark.serializerorg.apache.spark.serializer.KryoSerializer \ streaming-job.jar4. 参数调优的底层逻辑与计算过程4.1 Executor 数量与内存的数学关系很多人调参靠感觉其实 Executor 配置是可以算出来的。我总结了一个简单的计算流程第一步确定集群可用资源。假设集群 10 个节点每节点 32 核 128GYARN 可用比例 80%则总可用资源为 10 × 25 核 250 核10 × 100G 1000G。第二步确定每个 Executor 的配置。根据作业类型选计算密集型4~5 核内存按每核 2~4G 配。内存密集型大量缓存、Join2~3 核内存按每核 6~8G 配。混合型4 核每核 4G即 16G。第三步计算 Executor 数量。以 4 核 16G 为例按核数算250 / 4 62 个。按内存算1000 / (16 1.6) ≈ 56 个1.6 是 memoryOverhead。取较小值再留一点余量最终设 50 个左右。第四步计算并行度。总核数 50 × 4 200。spark.default.parallelism设为 200 × 2 400。spark.sql.shuffle.partitions根据 Shuffle 数据量单独算。这套流程看起来简单但实际用的时候要注意 YARN 队列的maxContainer限制。有些队列限制了单个作业能用的最大容器数算出来的 Executor 数量可能超过限制这时候要么换队列要么减少 Executor 数量、增大每个 Executor 的配置。4.2 内存溢出OOM的三种类型与对策OOM 是 Spark 作业最常见的报错但 OOM 和 OOM 不一样得对症下药。Driver OOM报错信息里出现Driver或SparkContext。原因通常是collect()数据量太大、广播变量太大、或者 DAG 太复杂。对策增大--driver-memory避免collect()大结果集用take()或write代替。Executor OOM堆内存报错信息里出现Executor和java.lang.OutOfMemoryError: Java heap space。原因通常是单个 Task 处理的数据量太大、或者spark.memory.fraction设得太高导致 User Memory 不够。对策增大--executor-memory增加spark.sql.shuffle.partitions减少单 Task 数据量或者降低spark.memory.fraction。Executor OOM堆外内存报错信息里出现java.lang.OutOfMemoryError: Direct buffer memory或容器被 YARN 杀掉报错Container killed by YARN for exceeding memory limits。原因是spark.executor.memoryOverhead不够。对策增大spark.executor.memoryOverhead或者直接增大--executor-memory因为 overhead 是按比例算的。实操心得遇到容器被 YARN 杀掉的情况先看 YARN 的日志里容器实际用了多少内存。如果超出申请值不多比如 10% 以内调大spark.executor.memoryOverhead就行。如果超出很多那说明确实有内存泄漏或者数据倾斜得从代码层面解决。4.3 数据倾斜的识别与参数级缓解数据倾斜是另一个性能杀手。表现是大部分 Task 很快跑完少数几个 Task 跑得特别慢甚至 OOM。在 Spark UI 里看 Stage 的 Task 时间分布如果 max 和 median 差好几倍基本就是倾斜了。参数级别的缓解手段有限但有几个可以用spark.sql.adaptive.skewJoin.enabledtrueSpark 3.0 的自适应倾斜 Join 处理会自动把倾斜的 Partition 拆成多个小 Partition。spark.sql.adaptive.advisoryPartitionSizeInBytes自适应优化时每个 Partition 的目标大小默认 64MB。调小可以让倾斜 Partition 拆得更细。spark.shuffle.minNumPartitionsToHighlyCompress这个参数比较冷门用于控制 Shuffle 压缩的阈值。不过说实话参数只能缓解真正解决数据倾斜还得从代码层面入手——加盐、拆分、或者用 Map 端 Join 代替 Reduce 端 Join。5. 常见问题与排查技巧实录5.1 参数不生效先检查优先级这是最常见的问题明明在 spark-submit 里设了参数代码里没生效。原因通常是代码里用SparkConf或SparkSession.builder.config()又设了一遍代码里的优先级更高把命令行参数覆盖了。排查方法在代码里打印sparkConf.getAll()看看实际生效的值是什么。或者看 Spark UI 的 Environment 页面里面列出了所有生效的配置项。注意spark.sql.shuffle.partitions如果在代码里用spark.conf.set()设了会覆盖命令行参数。但spark.default.parallelism只能在SparkConf里设spark.conf.set()对它无效。5.2 Executor 起不来检查资源计算提交作业后卡在ACCEPTED状态或者报Container is running beyond physical memory limits通常是资源计算有问题。检查以下几点--executor-memoryspark.executor.memoryOverhead是否超过了 YARN 单容器最大内存限制yarn.scheduler.maximum-allocation-mb。--executor-cores是否超过了 YARN 单容器最大核数限制yarn.scheduler.maximum-allocation-vcores。--num-executors是否超过了队列的maxContainer限制。集群是否有足够的空闲资源。5.3 任务跑得慢从 Spark UI 找线索任务慢的原因很多但 Spark UI 基本能定位到大部分问题症状可能原因排查方向Task 数量少每个 Task 数据量大并行度太低增大spark.sql.shuffle.partitionsTask 数量多每个 Task 很快并行度太高减小spark.sql.shuffle.partitions少数 Task 特别慢数据倾斜看 Task 输入数据量分布Shuffle Read 时间长网络或磁盘瓶颈调大spark.reducer.maxSizeInFlightGC 时间长内存不足或对象太多增大 Executor 内存用 Kryo任务卡在最后一个 Task数据倾斜或资源竞争检查是否有大 Partition5.4 常见问题速查表问题可能原因解决方案Driver OOMcollect 数据太大增大 driver-memory避免 collectExecutor OOM单 Task 数据太大增大 executor-memory增加分区数容器被 YARN 杀掉堆外内存超限增大 memoryOverhead参数不生效优先级冲突检查代码内配置Shuffle 拉取失败网络不稳定增大 maxRetries 和 retryWait任务跑得慢并行度不合理根据数据量调整分区数Executor 起不来资源不足或超限检查 YARN 配置和队列限制GC 频繁内存不足或序列化差增大内存改用 Kryo6. 我个人的调参习惯与最后分享调了这么多年 Spark 参数我最大的体会是不要一上来就调参先把代码逻辑和数据结构搞清楚。很多性能问题根本不是参数能解决的而是代码写得有问题——比如该用reduceByKey的地方用了groupByKey该用广播变量的时候没用该过滤的数据没提前过滤。参数调优是在代码已经合理的前提下做最后的性能压榨。我个人的习惯是新作业先用一套保守参数跑一遍看 Spark UI 里的 Stage 耗时、Task 分布、Shuffle 数据量、GC 时间然后针对性地调 1~2 个参数再跑一遍对比效果。不要一次调一堆参数那样根本不知道哪个参数起了作用。最后分享一个小技巧如果你不确定spark.sql.shuffle.partitions设多少合适可以先设一个较大的值比如 1000开启 AQE自适应查询优化让 Spark 自动合并小分区。这样即使初始值设大了Spark 也会自动优化到合适的大小。这个办法在 Spark 3.2 上特别好用基本可以告别手动调分区数的烦恼。还有一个参数我差点忘了提spark.sql.autoBroadcastJoinThreshold。默认是 10MB意思是小于 10MB 的表会自动广播到所有 Executor 做 Map 端 Join。如果你的小表刚好在 10MB 出头可以适当调大到 50MB 或 100MB能避免一次 Shuffle。但别调太大广播变量太大会撑爆 Driver 和 Executor 内存。这个参数和spark.sql.shuffle.partitions配合使用往往能带来意想不到的性能提升。