企业官网建设流程全解析

接手过不少 Spark 任务，我见过最多的问题不是集群资源不够，而是代码本身写得太随意。同样的数据量，一段规范代码可能 20 分钟跑完，换成一段只图“能出结果”的代码，拖到 2 小时还卡在某个 Stage 上。所谓 Spark 代码规范，并不是为了好看，而是在约束每一行代码对数据移动、内存占用、执行计划的影响。这篇博文围绕高性能 Spark 应用实践，把我在生产环境里沉淀下来的写代码习惯、参数配置、排查思路一起过一遍，适合正在写业务数仓任务、实时清洗逻辑或者算法特征加工的同学参考。

1. 先把性能问题的账算明白：为什么坏代码比小集群更致命

1.1 一个真实的对照：两段代码，两种命运

我在日常运维中做过一次对照。两份逻辑几乎一样的任务，都是处理 300GB 左右的用户行为日志，一份由某个临时脚本改造过来，另一份按项目规范重写。前者的做法是：读 JSON 后用 RDD 的 map 逐条解析，再 filter 掉无效字段，最后 groupByKey 做聚合；后者直接读成 DataFrame，提前过滤、只留需要的列，用 reduceByKey 或者 DataFrame API 完成聚合。两边都跑在同一个队列上，资源配额一模一样。

结果很典型：规范重写后的任务，整个 DAG 只有 6 个 Stage，Shuffle 数据量大约 40GB；原始脚本跑了 14 个 Stage，Shuffle 数据量超过 120GB，中间还有两次因为 Executor OOM 导致整个 Application 失败重试。最后耗时一个 18 分钟，一个 1 小时 50 分钟。这个差异完全不靠调参，纯粹是代码路径不同。

这就是我想强调的观点：性能优化不是等任务慢了才去翻 Spark UI，而是在写第一行代码时就考虑这条数据会被切成几份、哪些列会跨节点传输、聚合在哪一端发生。否则后面无论怎么加 Executor、调并行度，都是在给一个坏结构打补丁。

1.2 规范的本质：尽可能减少“数据搬家”

Spark 计算模型里，最贵的操作不是计算本身，而是数据在网络和磁盘之间的移动。这个概念我用生活里的事打比方：几个人一起做饭，菜在同一个厨房时，切菜、炒菜、装盘都很快；但如果要求每道菜都先送到另一个城市去加工再运回来，整个过程就会变得又慢又不可控。Spark 里的 Shuffle 就是这个“异地加工”。

具体来说，Shuffle 产生的原因主要是两类：一是按 Key 重新分区，比如 groupByKey、reduceByKey、distinct、join；二是通过 repartition 改变分区数量。每一次 Shuffle 都意味着父 Stage 的所有输出要先写磁盘，再被网络拉取到下一批 Task 里。磁盘 IO、网络 IO、序列化开销全在这里爆发。

所以代码规范的第一条原则可以用一句话概括：任何逻辑表达都要先问一句，这一步会不会引起全量数据重新洗牌，能不能在 Shuffle 之前把数据量砍掉一截。比如过滤条件能提前就提前，列能少选就少选，需要聚合的 Key 尽量在 map 端做一次预聚合。这不是“洁癖”，是在控制成本。

2. 数据抽象选错了，后面所有优化都是白做

2.1 DataFrame 优先，但 RDD 什么时候还要用

很多从老项目接手代码的人，第一反应就是 RDD 很灵活，什么都能干。确实，RDD 暴露了分区、依赖、计算函数，写起来很直接。但灵活的另一面是：Spark 的查询优化器很难介入 RDD 代码，它不知道你的 map 函数里到底读了哪些字段、哪些数据会被过滤掉，因此很多 RDD 版本的逻辑连最基本的“列剪枝”都享受不到。

我现在的默认原则是：能写 DataFrame / SQL 就绝不写 RDD。DataFrame 有 schema，Catalyst 优化器能看清楚数据的形状，能做谓词下推、列剪枝、常量折叠，甚至能在 join 时自动选择 Broadcast 还是 SortMerge。这份红利是白拿的。

RDD 也不是完全没有用。碰到三类场景我会退回 RDD：一是数据本身完全无结构，比如二进制报文、带自定义分隔符的杂散文本，强行套 schema 反而别扭；二是需要自己控制每个分区的处理粒度，比如某些模型打分要保证一批样本在同一个 Task 内完成；三是需要精准控制分区器的 hash 规则，避免默认分区把热点 Key 打散成奇怪的分布。除此之外，能用 DataFrame 就 DataFrame。

2.2 Catalyst 优化器到底替你省了哪几步

Catalyst 优化器做的事情，简单拆解有四步：逻辑计划解析、逻辑计划优化、物理计划生成、代码生成。其中逻辑优化阶段的谓词下推和列剪枝，对性能的影响最直白。

举个例子。有一张订单表和一张用户表做 join，最后只要用户维度的等级字段和订单金额。如果你先 join 再 filter 用户状态，优化器会自动尝试把 filter 下推到 scan 阶段，让每个分区只读需要的行。如果你在 DataFrame 上做了 select，只保留三列，那么 scan 时就不会再去解析和传输剩下几十个字段。这些优化在 RDD 版本里都不会发生。

但要注意，Catalyst 不是万能的。有些优化受限于表达方式。比如在 DataFrame API 里写一个自定义 UDF，优化器就看不清 UDF 内部逻辑，既不能下推也不能剪枝，整列数据都会被拉进函数里做序列化和反序列化。能用内置函数解决的问题，尽量别造自定义 UDF，这是 DataFrame 时代最容易踩的隐性坑。

2.3 一个 join 写法的执行计划对比

我实际演示过两种 join 写法。数据是两张分布表，一张 2 亿行，一张 3 万行的小表。第一种写法直接：

val result = bigDF.join(smallDF, Seq("key"))

没加任何提示，Spark 默认通过自动 Broadcast Join 阈值处理，执行计划里出现 BroadcastHashJoin，没有 Shuffle。第二种写法是同一个 join，但有人在 join 前对两张表做了 repartition(2000)，这一下就把 Broadcast 计划打破了，物理计划变成 SortMergeJoin，额外增加了一次全量 Shuffle，跑了 8 分钟。而第一种只要 30 秒。

这个案例说明：很多性能问题不是“不会优化”，而是无意间破坏了优化器的可优化条件。频繁 repartition、过多的 select 嵌套、把数据写成中间表再读回来，都会让优化器束手束脚。写完一段 SQL 或 DataFrame 链式操作后，应该养成一个习惯：explain() 看一眼物理计划，确认 join 是 Broadcast 还是 SortMerge，filter 是不是真的下推到了 scan。这几秒钟的检查，比在集群上跑完一遍再猜快得多。

3. 逐行拆解 Spark 代码规范：从分区、join 到算子选型

3.1 分区设计：repartition 与 coalesce 怎么选

分区数量直接决定 Task 数量和并行度。太大的分区数会导致调度开销暴涨，整个集群出现大量只有几 MB 输出的空 Task；太小则浪费并行能力，单 Task 处理时间过长。我比较常用的估算公式是：目标分区数 ≈ 总输入数据量 / 期望单 Task 处理数据量。日志类清洗任务里，单 Task 处理 128MB 到 256MB 比较合理，复杂聚合可以放宽到 500MB 左右。

repartition 和 coalesce 的差别，很多人只记住了“一个会引起 Shuffle，一个不会”，但没有深挖。coalesce 只能减少分区，而且是尽量把同一节点上的分区合并，天然能减少数据跨节点传输；repartition 可以增加也可以减少分区，shuffle 一定会发生。我在写最终结果到 Hive 表时，几乎只用 coalesce 控制输出文件数量，而不会在前面随意 repartition。如果并行度不够需要提高分区数，那只能用 repartition，这是没办法绕开的代价。

有一个容易踩的坑：coalesce 之后紧跟一个 map 类算子，很可能会打破 coalesce 的窄依赖结构，导致 Spark 重新做一次 Shuffle。这个现象在 Spark UI 里会看到明明写了 coalesce，Stage 数量却比你预期多。真要控制最终输出文件数，应该把 coalesce 放在所有转换的最后一步，或者直接在 write 前用 coalesce，不要在它后面再接复杂的 map。

3.2 join 策略：小表广播不是唯一招数

广播变量是最常见的优化手段之一。把几十 MB 的维度表做成广播，可以让每个 Executor 在本地保留一份完整数据，join 时完全不走 Shuffle。阈值参数是 spark.sql.autoBroadcastJoinThreshold，默认 10MB，生产上我建议根据 Executor 内存适量调大到 20MB 或 50MB，但别无限调大，否则每个 Executor 的堆内都要存一份，内存反而吃紧。

手动指定广播也有技巧，不要只依赖自动阈值。大表 join 小表时，可以在 SQL 里写：

SELECT /*+ BROADCAST(dim) */ * FROM fact f JOIN dim d ON f.key = d.key

DataFrame API 则用：

factDF.join(broadcast(dimDF), "key")

除了 join，广播变量还广泛用在“遍历维度做映射”的场景。比如一个 10 万条目的大字典，如果放在 mapPartitions 里逐条查询造成重复序列化，性能会很差。正确做法是用 SparkContext.broadcast 把字典打到每个 Executor，然后在 UDF 或 mapPartitions 里直接引用 broadcast .value，省掉每一次 shuffle 和网络查询。

另一个思路是不用广播而是做 Bucket Join。如果两张表都按同一个 Key 预先分桶成相同数量的文件，join 时可以做 Bucket 局部合并，连 Shuffle 都省掉。代价是建表和写入时需要稳定的分桶操作，属于数仓建模阶段就要设计的方案，不适合在临时任务里临时抱佛脚。

3.3 算子选型：groupByKey 是反面教材，mapPartitions 是宝藏

RDD 算子选型是最能拉开代码差距的地方。groupByKey 为什么是反面教材？因为它的逻辑是“把所有相同 Key 的 value 全量洗牌到一个分区，再统一处理”。哪怕你已经知道每个 Key 的求和值非常小，它也会把每个原始 record 全部搬一遍。

reduceByKey 则完全不同：它在 map 端先对每个分区内部的相同 Key 做一次预聚合，Shuffle 时传输的已经是中间聚合结果。一个计算 PV/UV 的任务，原始数据 100GB，预聚合后可能只有 1GB，Shuffle 量直接差两个数量级。代码上只是从 groupByKey().mapValues(.sum) 改成 reduceByKey(+ _)，执行计划却完全两样。

另一个高频场景是“对每个分区做批量初始化”。比如每台 Executor 上要加载一个模型文件、建立数据库连接，很多人会用 map 每条记录初始化一次，连接数被放大到百万级。正确做法是 mapPartitions，让每个 Task 只初始化一次，再在循环里复用。写数据库时同理，foreachPartition 配合连接池，比每条记录 foreach 里创建连接稳定得多。

我见过比较无语的代码是 map 里创建了 SimpleDateFormat，每条数据都 new 一次对象，导致 GC 压力巨大。这类对象应该放到 mapPartitions 外面统一初始化，或者直接用线程安全的 DateTimeFormatter 常量。一个看似无关紧要的算子选择，最后可能换来几十分钟的 GC 耗时。

3.4 序列化与内存：两个最容易忽略的配置项

Spark 默认的 Java 序列化器虽然稳定，但生成的字节数组特别膨胀，序列化速度也慢。Kryo 序列化在压缩率和速度上都明显占优，不过它需要提前注册类，否则也会退化到不可控状态。我的日常配置是：

spark.serializer org.apache.spark.serializer.KryoSerializer spark.kryo.registrationRequired true spark.kryo.registrator com.example.MyRegistrator

如果使用 DataFrame API，很多序列化工作被 Catalyst 优化成内部二进制格式，对这种配置的敏感度会低一些；但涉及 RDD、UDF、广播变量或者自定义对象时，Kryo 带来的收益非常可观。

内存上要理解 Spark 的 Unified Memory 模型。Executor 堆内内存被切成两块：Execution 内存和 Storage 内存，由 spark.memory.fraction 控制可用比例，默认 0.6，另外 0.4 是保留区，主要用于记录内部对象和防止 OOM。如果代码里频繁 cache 大表，又同时跑大 shuffle，两者会互相挤压。我一般建议不要盲目调大 spark.memory.fraction，因为保留区太小会让 JVM 在内存分配上变得脆弱，遇到极端情况反而更容易崩。

真正容易忽略的是 spark.executor.memoryOverhead。它不属于堆内 Execution 内存，但包含了 JVM 线程开销、NIO buffer、内部元数据。很多人的 OOM 问题不是堆内不够，而是 overhead 没给够。默认值是 executor 内存的 10%，碰上大规模 shuffle、大量 UDF 或者使用 Kryo 时，我通常手动配置成 1GB 甚至 2GB。这个参数在 Spark UI 里会显示为 “Memory Overhead”，被很多排查的人一眼略过。

4. 参数调优与执行计划：让规范落地到真实集群

4.1 内存与并行度参数的计算套路

调参先要有个估算过程，不是抄一个生产环境的配置就完事。假设一个 Executor 有 8GB 堆内内存，4 个核。通常我不会把每个 Executor 的内存拉满，而是留出 spark.executor.memoryOverhead 的空间。实际可用内存可以按堆内 × spark.memory.fraction 估算，也就是 8GB × 0.6 ≈ 4.8GB 的 Unified 区域。这个区域内既要放 shuffle 缓存，又要放 cache 数据，还要运行聚合的 HashMap。

并行度设置可以参考总核心数。比如集群有 20 个 Executor，每个 4 核，总并行 Task 数就是 80。那么 Spark 中一个 Stage 的理想 Task 数量最好在 80 到 160 之间，太少会浪费空闲核，太多则每个 Task 处理数据量过小、调度成本变大。尤其要注意 spark.sql.shuffle.partitions，默认值 200，很多任务在数据量只有几百 MB 时也强行切成 200 个分区。此时每个分区只有几 MB，网络和序列化开销占比非常高。我习惯根据数据量和并行度重新设它，一般不超过集群并发核数的 2 到 3 倍。

参数并不是越多越好。我见过有人把几十个 Spark 参数写进同一套提交脚本，然后发现照样慢。原因是很多参数是互相耦合的，比如 Executor 内存加大后，默认并行度没变，每个 Task 能吃到的内存更多，但 CPU 核数没增加，整体吞吐量不会有质变。做参数调整，最好一次只改一个变量，用 Spark UI 里各个 Stage 的耗时来做前后对比。

4.2 动态分配与 AQE：自动优化能救多少代码问题

Spark 3.0 之后的 Adaptive Query Execution（AQE）是我现在默认开启的东西。它能在运行时根据已经完成的 Stage 统计，自动调整后续的 Shuffle 分区数、自动转换 SortMergeJoin 为 BroadcastHashJoin、自动处理 Join 时的数据倾斜。配置如下：

spark.sql.adaptive.enabled true spark.sql.adaptive.coalescePartitions.enabled true spark.sql.adaptive.skewJoin.enabled true

AQE 解决了我之前手动调参的一半工作。以前要反复试 200、400、800 个 Shuffle 分区，现在它能根据 map 端的输出大小自动计算合适的 reduce 分区数。以前遇到大表 join 小表，偶尔因为阈值判断问题走了 SortMergeJoin，现在它能在 shuffle 过程中发现小表足够小，动态改成 Broadcast 策略。

但不要认为开启 AQE 就可以对代码完全放飞。AQE 主要优化的是 Shuffle 之后的 Reduce 端，它没办法弥补 map 端被 filter 过滤掉的行数、无法消除用户写出来的重度 UDF、也无法替代“尽量少读无用列”的自觉。它更像是安全网，可以兜底一部分不合理的并行度选择，但不能替代规范本身。

4.3 从 Spark UI 复盘一段慢任务的完整思路

慢任务排查时，我通常遵循一条固定路线：先看 Job 列表里哪些 Stage 耗时最长，认准大 Shuffle 发生的位置；然后点进 Stage 看 Shuffle Read Size / Shuffle Write Size，如果 Write 比输入还大，说明在算子选择上产生了大量中间数据；最后看 Event Timeline 里的 Executor 分布，判断是不是数据倾斜导致少数 Task 特别慢。

有一次我定位一个“波浪形”运行的任务，从 UI 上看到某个 Stage 的 200 个 Task 里，199 个只用了 5 秒，剩下 1 个跑了 40 分钟，Shuffle Read 接近全量数据。这种一眼就能断定是 Key 倾斜。如果代码里用 DataFrame API，可以通过 explain 看到实际执行计划，再针对倾斜 Key 做处理。

我还特别喜欢看 GC Time。某个 Stage 的 GC 时间超过总运行时间 20% 时，基本可以断定代码创建了过多短生命周期对象。此时不一定要调大内存，而应该检查是不是在 map 里频繁造对象、用了不合理的 UDF、或者把大数据集 cache 到了 Storage 区域导致 Execution 内存被挤压。UI 上的指标不是用来截图汇报的，是用来做推理的证据。

5. 生产环境常见问题与排查技巧实录

5.1 归因 OOM：先弄清楚是谁的内存

OOM 可以说是 Spark 任务里最常见的失败原因。但“OOM”只是一个结果，归因错误就会白调参。我会先看日志里的异常类：如果是 java.lang.OutOfMemoryError: Java heap space，说明堆内内存不足；如果是 Unable to acquire X bytes of memory，多半是堆外内存或 Off-heap 不足；如果是 Direct buffer memory，通常是 NIO 或 Kryo 的堆外缓冲问题。

堆内 OOM 常见于 reduce 端聚合过猛、collect() 拉取全量结果、cache 数据量超过 Storage 区域。很多人以为加大 spark.executor.memory 就行，但如果问题的根源是代码里有个 groupByKey 把所有数据都拉到单个分区，加内存只是延缓爆掉的时间。我的原则是：OOM 出现时先去看 Stage 的 Shuffle Read 和记录数，如果某 Task 拿到的是其他 Task 几百倍的输入，优先处理倾斜；如果整体都很大，再考虑调内存。

还有一个隐蔽问题：driver 端 OOM。当代码里执行了 collect() 把几亿行数据收回到 driver，或者大量使用 broadcast 变量列表，driver 的默认 1GB 内存很容易炸。这种情况调 Executor 内存毫无意义。规范上要避免 collect 大数据集，如果只是抽样观察数据，用 take(100) 或者 limit 就够了。

5.2 数据倾斜：定位、拆分、两阶段聚合

数据倾斜的经典场景是 join 和 groupBy。定位方法可以直接看 Spark UI：某个 Stage 里 Task 耗时严重不均，个别 Task 的 Input 或 Shuffle Read 远高于中位数。确认是某个 Key 倾斜后，最直接的手段是两阶段聚合。

两阶段聚合的思路是：第一次为 Key 加上随机前缀，打散到多个分区做部分聚合；第二次去掉前缀，再对真实 Key 做全局聚合。用 Scala 实现大致是：

rdd.map { case (key, value) => val salt = Random.nextInt(10) ((key, salt), value) }.reduceByKey(_ + _) .map { case ((key, _), sum) => (key, sum) } .reduceByKey(_ + _)

前缀范围不宜太大，10 到 100 之间可以接受。太大反而会造成第一轮 Stage 压力过大。对于 join 倾斜，可以先用 filter 把倾斜 Key 单独拆出来，加上随机前缀后与右表膨胀副本 join，再和正常 Key 的 join 结果合并。Spark 3 的 AQE 里已经内置了 skewJoin 优化，但前提是开启后且满足阈值，复杂场景仍然需要手动拆。

这个方案的关键是理解成本：两阶段聚合不会消除计算量，只是把原本集中在同一个 Task 上的压力分散到多个 Task，让每个 Task 都能吃到集群的并行能力。它不能解决单 Key 的总数据量超过单机内存的根本问题，所以有时候还要配合过滤掉低频噪声数据。

5.3 小文件问题：写入端管不住，读端就要哭

小文件是一个非常隐蔽但持续影响性能的问题。Hive 表如果写入时分区内出现几百乃至上千个小文件，后续读取时每个文件都会至少起一个 Task，元数据开销、打开文件耗时、任务调度开销全被放大。这种问题往往不是优化一段代码就能修复，它会拖累所有读取这张表的任务。

写入端最常见的做法是写入前 coalesce 到合理分区数。比如最终要写入一个分区的 2GB 数据，期望单文件 256MB，那么目标文件数就是 8 个左右。Spark 写入时可以用：

df.coalesce(8).write.mode("overwrite").format("parquet").saveAsTable("xxx")

如果是动态分区写入，每个分区都可能有分布不均匀的数据，直接把所有分区压到过少的分区又会产生大文件，因此需要结合 AQE 的动态分区裁剪和写入端 coalesce 参数来处理。一般我会开启 spark.sql.adaptive.coalescePartitions.enabled，并且检查每个分区的数据量是否与期望文件数匹配。测试环境里看起来不痛不痒的小文件问题，生产环境累积一个月后会产生巨大的读放大。

我也遇到过写入端压力不大，但 Hive 表本身已存在大量小文件的存量问题。这时候只能做一次合并重写，把分区里的小文件读出来再 coalesce 写回同等数据。这种操作既花时间又耗资源，所以更应该在规范上堵住源头：任何覆盖写表的任务，提交前先算一遍目标文件数。

5.4 一张问题排查速查表

我整理了一份常用的排查对照表，能覆盖大部分日常问题：

这张表不能替代经验判断，但适合作为第一步的入口。排查问题时不建议一次改三个参数，最好一次只改一个，再跑一次看 UI 变化，否则出了问题都不知道是哪个变量引起的。

6. 这些规范落地时的个人体会

我从这些年的 Spark 开发里得到的一条最重要的体会是：高性能代码不是写出来的，是“审”出来的。团队里定规范容易，执行起来难。后来我们在每次代码评审时把重点从“结果对不对”扩展为“结果对之外还要看执行计划”，每个人提交前都要贴一份 explain 关键输出，评审的人先看有没有大 Shuffle、有没有能提前的过滤、有没有不该存在的 repartition。这样几轮下来，新同学也能很快建立起对数据移动的直觉。

还有一个细节想分享：cache 慎用。很多同学习惯对中间结果 cache 一下，以为能加速，但如果没有复用，cache 纯属白占内存，甚至可能导致后续 Shuffle 的内存不足。判断标准很简单，同一个 DataFrame 在 DAG 里被分支引用超过一次时才值得 cache，否则不要随便加。

最后说说“先优化代码还是先加资源”的取舍。我的建议是：出现性能问题时，先在现有资源下去看代码和 UI，找到明显瓶颈后做针对性修改；只有在确认代码已经相对规范、瓶颈来自数据量级本身时，才去申请资源扩缩容。资源能解决一时的问题，但规范不佳的任务明天会继续产生新的问题。把这几条原则带进日常迭代，才是 Spark 应用长期稳定高效的基础。