1. 为什么RDD算子值得系统梳理一遍
刚接触分布式计算那会儿，我对RDD算子的理解就是“一堆能对集合做操作的函数”，直到真正在生产环境里处理TB级数据，才发现算子选错一个，整个作业的运行时间能从十分钟变成两小时。RDD算子是分布式数据处理的基石，它定义了数据怎么切分、怎么流转、怎么聚合，直接决定了作业的执行效率和资源消耗。这篇文章适合已经写过一些分布式作业、但对算子内部机制和选型逻辑还不够清晰的朋友，也适合想从“能跑通”进阶到“跑得好”的开发者。我会把RDD算子按转换和行动两大类拆开，逐个讲清楚它们的行为特征、适用场景、性能陷阱，以及我在实际项目里踩过的坑。核心关键词就一个：RDD算子，围绕它把整个体系讲透。
RDD的全称是弹性分布式数据集，它的核心抽象是“不可变、可分区、可并行操作”的数据集合。算子就是作用在这个集合上的函数，分为转换算子和行动算子。转换算子返回一个新的RDD，是惰性的，不会立即执行；行动算子触发实际计算，返回结果或写入存储。理解这个惰性机制是理解所有算子的前提。很多人写作业时习惯把转换算子当成普通集合操作来用，结果发现数据量一大就内存溢出，根本原因就是没搞清楚哪些算子会触发shuffle、哪些算子会缓存中间结果。
我见过不少开发者，写作业时凭直觉选算子，比如该用reduceByKey的地方用了groupByKey，该用mapPartitions的地方用了map，结果性能差了好几倍。这不是危言耸听，groupByKey会把所有相同key的数据拉到同一个分区，不做预聚合，网络传输量巨大；而reduceByKey会在map端先做本地聚合，再传送到reduce端，数据量能减少一个数量级。这类选型差异在数据量小的时候看不出来，一旦上到生产规模，差距就非常明显。
所以这篇文章不是简单的算子列表，而是按“行为特征—适用场景—性能考量—实操示例”四个维度来拆解每个算子。我会用生活化的类比来解释shuffle、宽依赖、窄依赖这些概念，也会给出可以直接抄的代码片段和参数配置。不管你是刚入门还是已经写过几十个作业，应该都能从中找到之前忽略的细节。
2. RDD算子的整体分类与设计逻辑
2.1 转换算子与行动算子的本质区别
转换算子的核心特征是惰性求值。当你写rdd.map(x => x * 2)的时候，Spark并不会立即对数据做乘法，而是记录下“这个RDD需要经过一个map操作”这个信息，构建出一张血缘图。只有遇到行动算子，比如count()、collect()、saveAsTextFile()，才会真正触发计算。这种设计的好处是Spark可以在执行前做全局优化，比如合并多个窄依赖操作、调整分区策略、决定哪些中间结果需要缓存。
行动算子则是触发实际计算的开关。它会把血缘图提交给调度器，拆分成多个Stage，每个Stage内部是一串可以流水线执行的窄依赖操作，Stage之间则是宽依赖，需要shuffle。理解这一点非常关键，因为shuffle是分布式计算中最昂贵的操作，涉及磁盘IO、网络传输和数据序列化。一个作业里shuffle次数越少、每次shuffle的数据量越小，整体性能就越好。
窄依赖和宽依赖的区分标准是：父RDD的每个分区是否只被子RDD的一个分区使用。map、filter、flatMap、mapPartitions这些是窄依赖，父分区和子分区是一对一或多对一的关系，不需要跨节点传输数据。groupByKey、reduceByKey、join、repartition这些是宽依赖，需要把数据按照key重新分发到不同节点，触发shuffle。宽依赖是Stage的边界，也是性能优化的重点区域。
2.2 算子选型的核心原则
第一条原则是能不用shuffle就不用shuffle。比如去重操作，distinct()底层是map加reduceByKey再加map，涉及shuffle；而如果只是想在每个分区内去重，可以用mapPartitions配合本地集合去重，完全避免网络传输。再比如排序，sortBy()是全局排序，需要shuffle；如果只需要分区内有序，用mapPartitions加本地排序就够了。
第二条原则是优先使用带预聚合的算子。reduceByKey和aggregateByKey都会在map端先做本地聚合，再发送到reduce端，而groupByKey不会。同样是对key做聚合，reduceByKey的网络传输量通常只有groupByKey的几分之一甚至几十分之一。combineByKey是最底层的聚合算子，可以自定义聚合逻辑，灵活性最高，但需要手动指定三个函数：创建累加器、分区内合并、分区间合并。
第三条原则是控制分区数量。分区太少，并行度不够，集群资源利用不充分；分区太多，任务调度开销大，小文件问题严重。一般来说，分区数设置为集群CPU核心数的2到4倍比较合适。repartition和coalesce都可以调整分区数，但repartition会触发shuffle，coalesce默认不触发shuffle，只能减少分区数。如果要从少变多，必须用repartition；如果只是合并小分区，用coalesce更高效。
2.3 血缘关系与容错机制
RDD的容错靠的是血缘关系，而不是数据副本。每个RDD都记录了自己是如何从父RDD计算得来的，一旦某个分区丢失，Spark可以根据血缘关系重新计算该分区，而不需要从头开始整个作业。这种设计大大降低了存储开销，但也意味着血缘链不能太长，否则恢复成本会很高。如果血缘链很长，建议在中间环节调用checkpoint()或cache()，把中间结果持久化下来，切断血缘链。
cache和persist的区别在于存储级别。cache默认使用MEMORY_ONLY，把数据以反序列化的形式存在内存里；persist可以指定MEMORY_AND_DISK、DISK_ONLY、MEMORY_ONLY_SER等多种级别。如果内存不够，MEMORY_ONLY会导致分区被丢弃，下次用到时重新计算；MEMORY_AND_DISK会溢写到磁盘，避免重算但增加IO开销。选择哪种级别，取决于数据量大小、内存资源和重算成本。我一般会先估算数据量，如果内存能放下就cache，放不下就用MEMORY_AND_DISK_SER，序列化后能省不少空间。
3. 转换算子逐个拆解与实操要点
3.1 单元素类算子：map、filter、flatMap
map是最基础的转换算子，对每个元素应用一个函数，返回一个新的RDD。它的特点是分区数不变，元素数量不变，一对一映射。比如rdd.map(x => x._2)提取二元组的第二个元素，或者rdd.map(x => (x, 1))把每个元素转成键值对。map的操作会在每个分区内独立执行，不需要跨节点传输数据，是窄依赖。
filter对每个元素应用一个布尔函数，保留返回true的元素。它的分区数不变，但元素数量可能减少。filter的一个常见陷阱是过滤后分区数据不均匀，某些分区可能变得很小甚至为空，导致后续操作并行度下降。如果过滤条件很苛刻，建议先filter再coalesce合并小分区，或者用repartition重新分布数据。
flatMap对每个元素应用一个返回迭代器的函数，然后把所有迭代器的结果展平。它和map的区别是：map是一对一，flatMap是一对多。比如rdd.flatMap(line => line.split(" "))把每行文本拆成单词，一个元素变成多个元素。flatMap常用于分词、展开嵌套结构、生成组合等场景。注意flatMap返回的迭代器不能为null，否则会报空指针异常。
这三个算子的共同点是都属于窄依赖，不会触发shuffle，可以流水线执行。Spark会把连续的map、filter、flatMap合并成一个Stage，每个分区独立处理，效率很高。我通常会把复杂的转换逻辑拆成多个小算子串联，而不是写一个巨大的map函数，这样代码可读性更好，Spark的优化器也更容易做谓词下推和列裁剪。
3.2 分区类算子：mapPartitions、mapPartitionsWithIndex、foreachPartition
mapPartitions和map的区别在于处理粒度。map是逐个元素处理，mapPartitions是把整个分区的迭代器交给函数，函数返回一个新的迭代器。这样做的好处是可以做分区级别的初始化，比如建立数据库连接、加载模型文件、打开文件句柄。如果每个元素都建一次连接，开销会非常大；用mapPartitions，每个分区只建一次连接，处理完整个分区的数据后再关闭。
mapPartitionsWithIndex比mapPartitions多了一个分区索引参数，可以在函数里知道当前处理的是第几个分区。这个算子常用于调试和分区级别的定制化处理，比如只对特定分区做特殊逻辑，或者给数据打上分区标签。我曾在处理倾斜数据时用这个算子识别出哪些分区数据量特别大，然后针对性地做加盐处理。
foreachPartition是行动算子，不是转换算子，但和mapPartitions配合使用很常见。它不返回RDD，而是对每个分区的迭代器执行操作，通常用于写入外部存储，比如把数据批量写入数据库、发送到消息队列、写入文件系统。和foreach相比，foreachPartition的粒度更粗，但批量操作的效率更高。比如写入数据库时，foreach是一条一条插入，foreachPartition可以攒一批再批量插入，性能差距可能达到几十倍。
使用mapPartitions类算子时要注意内存问题。因为函数接收的是整个分区的迭代器，如果函数内部把迭代器转成列表再处理，相当于把整个分区的数据加载到内存里。分区数据量大的时候，很容易OOM。正确的做法是流式处理迭代器，用iterator.map、iterator.filter这些方法，避免一次性加载全部数据。
3.3 键值对聚合类算子：reduceByKey、groupByKey、aggregateByKey、combineByKey
reduceByKey是最常用的聚合算子，对相同key的value做归约操作。它的执行过程分两步：先在map端做本地聚合，把相同key的value合并成一个值；然后把本地聚合结果按照key分区发送到reduce端，再做一次全局聚合。因为map端预聚合大大减少了网络传输量，reduceByKey的性能通常比groupByKey好很多。比如rdd.reduceByKey(_ + _)可以对相同key的value求和。
groupByKey把相同key的所有value收集到一个迭代器里，不做预聚合。这意味着所有数据都要经过网络传输，数据量大的时候网络和内存压力都很大。groupByKey的适用场景是确实需要拿到所有value的完整列表，比如要计算中位数、要排序、要做复杂的自定义聚合。如果只是求和、求平均、求最大值，用reduceByKey或aggregateByKey更合适。
aggregateByKey比reduceByKey更灵活，可以指定初始值、分区内聚合函数和分区间聚合函数。比如求平均值，可以先用aggregateByKey把value累加成(sum, count)的二元组，然后再用map做除法。aggregateByKey的初始值会在每个分区内使用一次，所以初始值必须是中性值，不能影响聚合结果。比如求和时初始值用0，求积时初始值用1。
combineByKey是最底层的聚合算子，reduceByKey和aggregateByKey底层都是用它实现的。它需要三个函数：createCombiner负责把第一个value转成累加器，mergeValue负责把后续value合并到累加器，mergeCombiners负责合并不同分区的累加器。combineByKey的灵活性最高，可以实现任意复杂的聚合逻辑，但代码也最繁琐。我一般只在reduceByKey和aggregateByKey无法满足需求时才用它。
| 算子 | 是否预聚合 | 网络传输量 | 适用场景 |
|---|---|---|---|
| reduceByKey | 是 | 小 | 求和、求最大值、求最小值 |
| groupByKey | 否 | 大 | 需要完整value列表 |
| aggregateByKey | 是 | 小 | 求平均、自定义聚合 |
| combineByKey | 是 | 小 | 复杂聚合逻辑 |
3.4 连接类算子：join、leftOuterJoin、rightOuterJoin、fullOuterJoin
join用于把两个键值对RDD按照key连接起来。它的执行过程是：先对两个RDD按照key做分区，把相同key的数据拉到同一个分区，然后在该分区内做笛卡尔积。join是宽依赖，会触发shuffle。如果两个RDD的分区器相同，并且其中一个RDD已经按照key分区，那么join可以避免shuffle，直接在每个分区内做本地连接。
leftOuterJoin保留左RDD的所有key，右RDD没有匹配的用None填充。rightOuterJoin反过来，保留右RDD的所有key。fullOuterJoin保留两边的所有key，没有匹配的用None填充。这些外连接算子在处理不完整数据时很有用，比如左表是用户表，右表是订单表，leftOuterJoin可以找出所有用户及其订单，没有订单的用户订单字段为None。
join的性能陷阱主要是数据倾斜。如果某个key的数据量特别大，比如热门商品、默认值、空值，那么处理这个key的任务会成为瓶颈，其他任务早就完成了，它还在跑。解决数据倾斜的方法有：加盐打散、拆分热点key单独处理、用map端join代替reduce端join。map端join适用于一个小表和一个大表连接的情况，把小表广播到每个节点，大表直接在本地做连接，完全避免shuffle。
3.5 集合类算子：union、intersection、subtract、distinct
union把两个RDD合并成一个，不进行去重。它的分区数是两个RDD分区数之和，是窄依赖。union常用于合并多个数据源的结果，比如把多天的日志合并在一起。注意union不会去重，如果两个RDD有重复元素，结果里也会有重复。
intersection求两个RDD的交集，会去重。它的实现是先把两个RDD转成(key, null)的形式，然后做cogroup，最后过滤出两边都有的key。intersection涉及shuffle，性能开销较大。subtract求差集，返回在第一个RDD但不在第二个RDD中的元素，也会去重，同样涉及shuffle。
distinct对RDD去重，底层是map转成(key, null)再reduceByKey再map取key。distinct涉及shuffle，如果数据量很大，开销不小。如果只是想去重后计数，可以用countApproxDistinct做近似去重，性能好很多，但结果是近似值。如果数据量不大，也可以先collect到本地用集合去重，再并行化回去，但要注意内存限制。
3.6 重分区类算子：repartition、coalesce、partitionBy
repartition重新分区，可以增加或减少分区数，会触发shuffle。它的实现是给每个元素随机分配一个新分区，然后按照新分区做shuffle。repartition常用于增加并行度、解决数据倾斜、调整分区大小。比如从HDFS读取的文件分区数等于文件块数，可能很少，用repartition增加分区数可以提高并行度。
coalesce默认不触发shuffle，只能减少分区数。它的实现是把多个小分区合并成一个大分区，在同一个节点上完成，不需要网络传输。coalesce常用于过滤后合并小分区，减少任务调度开销。如果coalesce要增加分区数，必须设置shuffle=true，此时它等价于repartition。
partitionBy按照指定的分区器重新分区，只适用于键值对RDD。Spark内置了HashPartitioner和RangePartitioner，也可以自定义分区器。partitionBy常用于优化后续的join或聚合操作，如果两个RDD用相同的分区器分区，join时可以避免shuffle。比如先对两个RDD都调用partitionBy(new HashPartitioner(100))，然后再join，效率会高很多。
3.7 排序类算子：sortBy、sortByKey
sortBy对RDD元素排序，可以指定升序或降序。它的实现是先用RangePartitioner对数据采样分区，保证每个分区的数据范围不重叠，然后对每个分区内部排序。sortBy涉及shuffle，因为需要全局有序。如果只需要分区内有序，用mapPartitions加本地排序就够了，不需要shuffle。
sortByKey对键值对RDD按照key排序，底层也是RangePartitioner。sortByKey常用于需要按key有序输出的场景，比如输出到文件时希望按时间戳排序。注意sortByKey默认是升序，降序需要传false参数。排序操作的内存开销较大，因为RangePartitioner需要采样来确定分区边界，采样比例可以通过spark.sql.execution.sortBeforeRepartition参数调整。
4. 行动算子与持久化策略
4.1 常见行动算子：collect、count、take、first、reduce、foreach
collect把RDD的所有元素拉到Driver端，返回一个数组。它的风险是数据量大的时候Driver端内存溢出。我见过有人对TB级数据调collect，结果Driver直接崩了。collect只适合小结果集，比如聚合后的统计结果、配置数据、小维表。如果确实需要把大量数据拉到本地，应该先coalesce到一个分区，再逐批collect，或者直接写入分布式存储。
count返回RDD的元素个数，会触发全量计算。take(n)返回前n个元素，不会计算全量数据，只计算包含前n个元素的分区，效率比collect高。first返回第一个元素，等价于take(1)。reduce对RDD元素做归约，返回单个值，常用于求和、求最大值。foreach对每个元素执行操作，不返回结果，常用于写入外部存储或打印调试信息。
行动算子的选择要考虑结果集大小和计算成本。如果只是想知道数据量，用count；如果想看样本数据，用take；如果想做聚合统计，用reduce或aggregate；如果要写入外部系统，用foreach或foreachPartition。避免对大结果集用collect，这是最常见的OOM原因之一。
4.2 cache与persist的选型逻辑
cache和persist用于把RDD的中间结果持久化，避免重复计算。cache是persist的特例，使用MEMORY_ONLY级别。persist可以指定多种存储级别，包括MEMORY_ONLY、MEMORY_AND_DISK、MEMORY_ONLY_SER、MEMORY_AND_DISK_SER、DISK_ONLY等。选择哪种级别，取决于数据量、内存资源、重算成本和序列化开销。
MEMORY_ONLY把数据以反序列化的Java对象形式存在内存里，读取速度快，但占用内存大。如果内存不够，部分分区会被丢弃，下次用到时重新计算。MEMORY_ONLY_SER把数据序列化后存储，占用内存小，但读取时需要反序列化，增加CPU开销。MEMORY_AND_DISK在内存不够时溢写到磁盘，避免重算但增加IO开销。DISK_ONLY只存磁盘，适合数据量特别大、内存完全放不下的场景。
我的经验是：如果数据量能放进内存，用MEMORY_ONLY；如果放不下但重算成本高，用MEMORY_AND_DISK_SER；如果重算成本低，干脆不缓存，让Spark自己决定。缓存不是越多越好，缓存过多会导致内存不足，反而拖慢整体性能。一般只缓存会被多次使用的RDD，比如迭代计算中的中间结果、被多个行动算子共享的RDD。
4.3 checkpoint的使用场景
checkpoint把RDD的数据写入可靠的分布式存储，比如HDFS，切断血缘链。和cache不同，checkpoint的数据在作业结束后仍然存在，可以跨作业使用。checkpoint适用于血缘链特别长、重算成本极高的场景，比如迭代一百次的机器学习训练，每次迭代都依赖上一次的结果，血缘链长达上百个RDD，一旦某个分区丢失，重算成本无法接受。
使用checkpoint需要先设置存储目录，sc.setCheckpointDir("hdfs://...")，然后对RDD调用checkpoint()。注意checkpoint是惰性的，需要遇到行动算子才会真正执行。而且checkpoint会触发一次额外的计算，把数据写入存储，所以最好在checkpoint之前先cache，避免重复计算。checkpoint和cache可以配合使用：先cache，再checkpoint，这样第一次计算后数据既在内存里又在磁盘上，后续使用优先读内存，内存不够时读磁盘。
4.4 持久化与检查点的组合策略
在实际项目中，我通常这样组合：对于迭代计算，每轮迭代的中间RDD先cache，每十轮checkpoint一次。这样大部分时候读内存，偶尔读磁盘，血缘链也不会太长。对于被多个行动算子共享的RDD，cache起来避免重复计算。对于一次性使用的RDD，不缓存，让Spark按需计算。
还有一个技巧是用unpersist()及时释放不再需要的缓存。Spark的缓存是LRU策略，但主动释放可以更早腾出内存。比如一个RDD在某个阶段用完后，后面不再需要，就调用unpersist()把它从内存中移除。我见过有人缓存了几十个RDD，内存爆了还不知道为什么，其实就是忘了释放。
5. 常见问题与排查技巧实录
5.1 数据倾斜的识别与处理
数据倾斜的表现是：大部分任务很快完成，少数任务跑得特别慢，甚至OOM。在Spark UI上可以看到某些任务的shuffle read size远大于其他任务，或者某些任务的执行时间远长于平均值。数据倾斜的根因是某些key的数据量远大于其他key，比如空值、默认值、热门商品ID。
处理数据倾斜的方法有几种。第一种是加盐打散：给每个key加上随机前缀，把一个大key拆成多个小key，聚合完再去掉前缀。比如rdd.map(x => (Random.nextInt(10) + "_" + x._1, x._2)).reduceByKey(_ + _).map(x => (x._1.split("_")(1), x._2)).reduceByKey(_ + _)。第二种是拆分热点key：把数据分成热点部分和非热点部分，分别处理后再合并。第三种是用map端join代替reduce端join：如果一个小表和一个大表连接，把小表广播到每个节点，大表直接在本地连接。
我遇到过一次严重的数据倾斜，某个key的数据量占了总量的80%，原因是上游数据里有个默认值没有过滤掉。后来在聚合前先filter掉这个默认值，问题就解决了。所以遇到数据倾斜，先检查数据质量，看看是不是有异常值、空值、默认值在作怪。
5.2 内存溢出与GC调优
内存溢出通常发生在collect、groupByKey、mapPartitions转列表这些操作上。collect是把所有数据拉到Driver，Driver内存有限，数据量大就崩。groupByKey是把相同key的所有value拉到内存，某个key的数据量特别大就崩。mapPartitions如果函数内部把迭代器转成列表，也会把整个分区的数据加载到内存。
解决内存溢出的方法：用reduceByKey代替groupByKey，用foreachPartition代替foreach，用流式处理代替列表处理，增加Executor内存，调整GC策略。GC调优方面，如果数据量大且频繁GC，可以用G1垃圾回收器，设置spark.executor.extraJavaOptions=-XX:+UseG1GC。如果序列化开销大，可以用Kryo序列化，设置spark.serializer=org.apache.spark.serializer.KryoSerializer，比Java序列化快很多。
5.3 任务执行缓慢的排查思路
任务执行缓慢的原因很多，按优先级排查：第一看数据倾斜，第二看shuffle量，第三看分区数，第四看序列化方式，第五看资源分配。数据倾斜前面讲过了。shuffle量大通常是算子选型问题，比如该用reduceByKey用了groupByKey。分区数太少导致并行度不够，分区数太多导致调度开销大。序列化方式影响网络传输和磁盘IO，Kryo比Java快。资源分配包括Executor数量、每个Executor的内存和CPU核心数，需要根据数据量和集群规模调整。
我一般会先看Spark UI的Stage页面，找出耗时最长的Stage，然后看这个Stage的shuffle read和shuffle write量，再看任务执行时间的分布。如果任务时间分布很不均匀，基本就是数据倾斜。如果所有任务都很慢，可能是分区数不够或者资源不足。如果shuffle量特别大，检查算子选型。
5.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 任务执行慢 | 数据倾斜 | 看任务时间分布 | 加盐、拆分热点key |
| Driver OOM | collect数据量太大 | 看Driver日志 | 改用take或写入存储 |
| Executor OOM | groupByKey或mapPartitions转列表 | 看Executor日志 | 改用reduceByKey或流式处理 |
| shuffle量大 | 算子选型不当 | 看Stage的shuffle量 | 用带预聚合的算子 |
| GC频繁 | 内存不足或序列化开销大 | 看GC日志 | 增加内存、用Kryo序列化 |
| 分区数不合理 | 并行度不够或调度开销大 | 看分区数和任务数 | 用repartition或coalesce调整 |
6. 算子组合实战：从需求到实现
6.1 词频统计的多种实现与性能对比
词频统计是分布式计算的Hello World，但不同实现方式的性能差异很大。最直观的写法是rdd.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)，这是标准写法，map端预聚合，性能好。如果写成rdd.flatMap(_.split(" ")).map((_, 1)).groupByKey().mapValues(_.sum)，功能一样但性能差很多，因为groupByKey不做预聚合，所有数据都要走网络。
还有一种写法是rdd.flatMap(_.split(" ")).countByValue()，countByValue是行动算子，返回一个Map。它的底层实现是先用map转成(key, 1)再reduceByKey再collect，所以结果集小的时候可以用，结果集大的时候Driver会OOM。我一般用reduceByKey的写法，既高效又安全。
如果数据量特别大，还可以用aggregateByKey做更复杂的统计，比如同时求词频和平均词长。rdd.flatMap(_.split(" ")).map(word => (word, (1, word.length))).aggregateByKey((0, 0))((acc, v) => (acc._1 + v._1, acc._2 + v._2), (acc1, acc2) => (acc1._1 + acc2._1, acc1._2 + acc2._2)).mapValues(x => (x._1, x._2.toDouble / x._1))。这样一次遍历就能算出词频和平均词长，比分开算两次效率高。
6.2 多表关联的算子选型
假设有两个RDD：用户RDD是(userId, userInfo)，订单RDD是(userId, orderInfo)。要找出每个用户及其订单，可以用join或leftOuterJoin。如果两个RDD都很大，用join会触发shuffle，性能取决于数据分布。如果用户RDD比较小，可以collect到本地做成广播变量，然后用map端join：orders.mapPartitions(partition => { val userMap = broadcastUsers.value; partition.flatMap(order => userMap.get(order._1).map(user => (user, order._2))) })。这样完全避免shuffle，性能提升非常明显。
如果两个RDD都很大且都需要参与连接，可以先用partitionBy把两个RDD按照相同的分区器分区，然后再join。这样join时不需要额外的shuffle，因为数据已经在正确的分区上了。val userPartitioned = users.partitionBy(new HashPartitioner(100)); val orderPartitioned = orders.partitionBy(new HashPartitioner(100)); userPartitioned.join(orderPartitioned)。注意partitionBy本身会触发一次shuffle，但后续的join就不需要了，如果join会被多次执行，这个开销是值得的。
6.3 迭代计算的缓存与检查点策略
机器学习训练、图计算这类迭代场景，每轮迭代都依赖上一轮的结果。如果不缓存，每轮都要从头计算，血缘链越来越长，计算成本指数级增长。正确的做法是每轮迭代的中间RDD都cache，每十轮checkpoint一次。var rdd = initialRDD; for (i <- 1 to 100) { rdd = rdd.map(update).cache(); if (i % 10 == 0) { rdd.checkpoint(); rdd.count() } }。注意checkpoint后要调用一次行动算子触发实际写入，否则checkpoint不会执行。
还有一个细节是checkpoint后要unpersist旧的缓存，释放内存。val oldRdd = rdd; rdd = rdd.map(update).cache(); oldRdd.unpersist()。这样内存里始终只有最近几轮的缓存，不会无限增长。我见过有人迭代一百轮缓存一百个RDD，内存直接爆掉，其实就是忘了释放旧缓存。
6.4 自定义分区器的实现与应用
Spark内置的HashPartitioner和RangePartitioner能满足大部分需求，但有些场景需要自定义分区器。比如要保证某些key的数据落在同一个分区，或者要按照key的某个范围分区。自定义分区器需要继承Partitioner类，实现numPartitions和getPartition两个方法。
class CustomPartitioner(numParts: Int) extends Partitioner { override def numPartitions: Int = numParts override def getPartition(key: Any): Int = { val k = key.asInstanceOf[String] if (k.startsWith("hot_")) 0 else Math.abs(k.hashCode) % (numParts - 1) + 1 } }这个分区器把hot_开头的key都分到0号分区，其他key均匀分到剩余分区。这样可以把热点数据集中处理，避免分散到多个分区导致每个分区都有热点。自定义分区器常用于优化join和聚合操作，让数据按照预期的方式分布。
6.5 算子链的调试与优化
Spark会把连续的窄依赖算子合并成一个Stage，每个分区流水线执行。但有些算子会打断流水线，比如shuffle、cache、checkpoint。调试算子链的时候，我会在关键节点调用rdd.toDebugString查看血缘图，确认算子是否按照预期合并。如果发现意外的shuffle，就检查是不是某个算子触发了宽依赖。
优化算子链的一个技巧是调整算子的顺序。比如先filter再map，filter可以减少数据量，map处理的数据就少了。先reduceByKey再filter，reduceByKey会触发shuffle，filter放在后面可以减少shuffle后的数据量。但filter放在reduceByKey前面可能更好，因为可以减少参与shuffle的数据量。具体怎么放，要看filter的选择性，如果filter能过滤掉大部分数据，放在前面；如果过滤性不强，放在后面。
还有一个技巧是用mapPartitions代替多个map和filter的组合。比如rdd.map(f).filter(g).map(h)可以合并成rdd.mapPartitions(_.map(f).filter(g).map(h))，减少函数调用开销。但这样会降低代码可读性，只在性能瓶颈时使用。
7. 我在实际项目中的算子选型心得
算子选型这件事，说到底是在正确性、性能和代码可维护性之间找平衡。我刚开始写分布式作业的时候，只关心能不能跑出正确结果，后来数据量上来了，才发现性能问题不解决根本跑不完。再后来维护别人的代码，又发现可读性太差，改一个逻辑要花半天理解。
我的经验是：先保证正确性，再优化性能，最后考虑可读性。但优化性能的时候不要过度，有些优化只在小数据量下有效，数据量大了反而适得其反。比如coalesce合并小分区，如果合并后分区数据量太大，反而容易OOM。再比如cache，如果缓存的数据后面只用一次，缓存的开销比重新计算还大。
还有一个心得是：不要迷信“最佳实践”，要根据自己的数据特征和集群环境做实验。同一个算子，在不同数据分布、不同集群配置下，性能表现可能完全不同。我一般会先用小数据集做对比实验，测出几个候选方案的执行时间和资源消耗，再选最优的。Spark UI是最好的老师，多看看Stage的shuffle量、任务时间分布、GC时间，比看多少文档都管用。
最后分享一个排查问题的习惯：遇到性能问题，先看Spark UI，再看日志，最后看代码。Spark UI能告诉你哪里慢，日志能告诉你为什么慢，代码能告诉你怎么改。顺序不要反，上来就改代码往往改不到点子上。
