1. 为什么 RDD 持久化是 Spark 性能优化的第一道分水岭很多人写 Spark 任务跑通之后就觉得万事大吉直到某天数据量从几百 MB 涨到几十 GB原本三分钟出结果的作业突然跑了四十分钟甚至直接 OOM 崩掉。你去翻日志发现某个 RDD 被反复计算了七八次每次都要从源头重新读数据、重新做转换。这时候你才意识到Spark 的惰性求值机制虽然优雅但它默认不帮你保存中间结果——每次遇到行动操作它就从血缘关系的最顶端重新算一遍。RDD 持久化就是解决这个问题的核心手段。说白了它就是把某个 RDD 的计算结果在第一次算出来之后缓存到内存或磁盘上后续再需要这个 RDD 的数据时直接拿缓存不用重新算。这个机制看起来简单但实际用起来坑不少缓存级别选错了等于没缓存缓存位置放错了反而拖慢速度该缓存的没缓存导致重复计算不该缓存的缓存了白白占内存。我见过太多团队在调优 Spark 任务时第一刀砍的就是持久化策略效果往往立竿见影。这篇文章面向的是已经能写 Spark 任务、但对性能调优还没有系统认知的开发者。我会从 RDD 持久化的底层逻辑讲起把 cache、persist、checkpoint 这几个容易混淆的概念彻底掰开然后给出不同场景下的选型建议和参数配置方法最后分享几个我在实际项目中踩过的坑和排查技巧。你不需要有调优经验但最好写过至少一个完整的 Spark 作业这样理解起来会更顺畅。2. RDD 持久化的核心机制与设计逻辑拆解2.1 惰性求值带来的重复计算问题Spark 的 RDD 转换操作map、filter、flatMap 等都是惰性的调用的时候并不会立即执行只是记录下“这个 RDD 是怎么从父 RDD 算出来的”这条血缘关系。只有遇到行动操作count、collect、saveAsTextFile 等时Spark 才会沿着血缘关系从后往前推导生成一个 DAG然后提交任务去执行。这个设计的好处是 Spark 可以在全局视角下做优化比如合并窄依赖的转换、调整分区策略。但坏处也很明显如果你对同一个 RDD 调用了多次行动操作每次都会触发一次完整的血缘回溯。举个例子你对一个经过复杂 ETL 清洗的 RDD 先做了一次 count 看数据量然后又做了一次 collect 取结果如果没有持久化整个清洗流程会被完整执行两遍。数据量小的时候无所谓数据量大的时候这两遍可能就是几十分钟的差距。更隐蔽的情况是迭代式计算。机器学习算法里经常需要反复扫描同一份数据比如梯度下降每一轮迭代都要用到全量训练集。如果每一轮都从头算时间复杂度直接翻倍。持久化的本质就是打断血缘链把中间结果物化下来让后续的计算直接从物化点开始。2.2 cache 和 persist 的本质区别很多人以为 cache 和 persist 是两个不同的东西其实 cache 就是 persist 的一个特例。看源码就知道cache()内部调用的就是persist(StorageLevel.MEMORY_ONLY)。所以从功能上讲cache 等价于用默认的 MEMORY_ONLY 级别做持久化。那为什么还要保留两个方法主要是语义上的区分。cache 表达的是一种“我就是要缓存用最常用的方式就行”的意图适合大多数场景。persist 则给了你精细控制存储级别的能力当你需要把数据存到磁盘、或者做序列化压缩、或者多副本存储时就必须用 persist 并显式指定 StorageLevel。这里有一个容易忽略的细节cache 和 persist 都是惰性的调用它们的时候并不会立即触发缓存只是给 RDD 打了一个标记。真正触发缓存是在第一次遇到行动操作、该 RDD 被计算出来的时候。所以如果你 persist 之后没有跟任何行动操作缓存是不会发生的。这一点新手特别容易搞错以为调了 persist 就万事大吉结果发现根本没缓存上。2.3 StorageLevel 的存储级别全解析StorageLevel 决定了数据缓存在哪里、以什么形式缓存、存几份。它的构造参数有五个维度useDisk、useMemory、useOffHeap、deserialized、replication。这五个布尔值或整数值组合起来形成了 Spark 预定义的十几种存储级别。最常用的几种我列一下存储级别含义适用场景MEMORY_ONLY只存内存不序列化存不下就丢弃数据量小、频繁访问、对速度要求极高MEMORY_AND_DISK优先存内存存不下的部分溢写到磁盘数据量中等、内存不够但不想重算MEMORY_ONLY_SER存内存但序列化节省空间但读取要反序列化内存紧张、数据量大、能接受一定 CPU 开销MEMORY_AND_DISK_SER内存序列化存储溢写磁盘数据量大、内存有限、追求空间利用率DISK_ONLY只存磁盘数据量极大、内存完全放不下、访问频率低OFF_HEAP存堆外内存需要减少 GC 压力、内存充足选择存储级别的核心权衡是内存 vs 磁盘、序列化 vs 反序列化、空间 vs 速度。MEMORY_ONLY 最快因为数据以 Java 对象形式直接存在内存里读取时零转换开销。但 Java 对象有巨大的内存开销一个包含几个字段的对象实际占用可能是原始数据的 3 到 5 倍。MEMORY_ONLY_SER 把数据序列化成字节数组再存内存占用能降到 1/3 左右但每次读取都要反序列化CPU 开销上去了。我的经验是如果数据量不大比如能轻松放进内存的 30% 以内直接用 MEMORY_ONLY简单粗暴效果好。如果数据量中等偏大用 MEMORY_AND_DISK_SER牺牲一点 CPU 换空间避免频繁溢写磁盘。如果数据量极大考虑 DISK_ONLY 配合合理的分区数或者干脆重新设计计算流程不要硬扛。2.4 checkpoint 与持久化的本质差异checkpoint 经常和 persist 一起被提到但它们的定位完全不同。persist 是临时的、运行时的缓存作业结束后缓存就释放了而且它保留完整的血缘关系一旦缓存丢失还能重算。checkpoint 则是把 RDD 的数据写到可靠的分布式文件系统上永久保存并且会切断血缘关系。为什么要切断血缘因为血缘链太长会带来两个问题一是容错恢复时重算代价太大二是序列化血缘关系本身会占用 Driver 内存。在迭代计算中如果每一轮都基于上一轮的结果继续构建血缘几十轮下来血缘链会变得极长这时候 checkpoint 就很有必要。checkpoint 的使用有个关键点它也是惰性的而且建议先 persist 再 checkpoint。因为 checkpoint 会触发一次完整的计算并把结果写出去如果不先 persist这次计算之后数据就丢了后续用到这个 RDD 时还要再算一遍。先 persist 的话checkpoint 写出的同时缓存也建立了后续直接读缓存。3. 持久化实操从代码到参数配置的完整落地3.1 基础用法与代码示例先看最基本的用法。假设我们有一个从日志文件读进来、经过清洗转换的 RDD后续要多次使用from pyspark import SparkContext, StorageLevel sc SparkContext(local[4], PersistenceDemo) # 模拟一个计算代价较高的 RDD raw_rdd sc.textFile(hdfs://path/to/large/logfile) cleaned_rdd raw_rdd \ .filter(lambda line: len(line) 0) \ .map(lambda line: line.split(\t)) \ .filter(lambda fields: len(fields) 5) \ .map(lambda fields: (fields[0], int(fields[3]))) # 方式一cache等价于 persist(MEMORY_ONLY) cleaned_rdd.cache() # 方式二显式指定存储级别 cleaned_rdd.persist(StorageLevel.MEMORY_AND_DISK_SER) # 第一次行动操作触发实际计算和缓存 count cleaned_rdd.count() print(fTotal records: {count}) # 第二次行动操作直接从缓存读取不会重新计算 top10 cleaned_rdd.takeOrdered(10, keylambda x: -x[1]) print(fTop 10: {top10}) # 用完记得释放 cleaned_rdd.unpersist()这段代码里有个细节值得注意unpersist()的调用。很多教程不讲这个但在长时间运行的应用里不及时释放缓存会导致内存被占满后续的 RDD 没有空间缓存甚至引发频繁 GC。我的习惯是在一个 RDD 确定不再使用后立即 unpersist尤其是在循环或迭代场景中。3.2 存储级别选择的量化依据怎么判断该用哪个存储级别不能凭感觉得算账。假设你的数据集大小是 D集群可用内存是 M序列化后的膨胀系数是 s通常 0.2 到 0.5 之间取决于数据类型和序列化器。如果 D M * 0.3用 MEMORY_ONLY。留 70% 内存给其他 RDD 和运行时开销。如果 M * 0.3 D M * 0.7用 MEMORY_ONLY_SER。序列化后实际占用约 D * s能塞进内存同时给其他数据留空间。如果 D M * 0.7用 MEMORY_AND_DISK_SER。内存放不下的部分溢写磁盘序列化减少内存压力。如果 D 远大于 M比如 D M * 3考虑 DISK_ONLY 或者重新分区后分批处理。这里的膨胀系数 s 怎么估最直接的办法是跑一个小规模测试用rdd.persist(StorageLevel.MEMORY_ONLY_SER)之后看 Spark UI 里 Storage 页签显示的 Size in Memory和原始数据大小对比。或者用rdd.getNumPartitions()和每个分区的平均记录数、每条记录的平均字节数来估算。3.3 checkpoint 的正确配置姿势checkpoint 需要先设置一个可靠的存储路径通常是 HDFS 路径# 设置 checkpoint 目录 sc.setCheckpointDir(hdfs://path/to/checkpoint/dir) # 构建一个血缘链较长的 RDD iterative_rdd sc.parallelize(range(1000000), 100) for i in range(50): iterative_rdd iterative_rdd.map(lambda x: x * 1.0001 i) # 先 persist 再 checkpoint iterative_rdd.persist(StorageLevel.MEMORY_AND_DISK_SER) iterative_rdd.checkpoint() # 触发计算 result iterative_rdd.reduce(lambda a, b: a b) print(fResult: {result})这里的关键顺序是 persist 在前、checkpoint 在后。checkpoint 本身会触发一次计算如果之前没有 persist这次计算的结果直接写到 HDFS内存里不留后续再用还得重算。先 persist 的话checkpoint 写 HDFS 的同时内存/磁盘缓存也建立了后续读取走缓存速度快得多。还有一个坑checkpoint 会切断血缘关系这意味着 checkpoint 之后的 RDD 如果出问题无法通过血缘回溯恢复只能从 checkpoint 数据重新加载。所以 checkpoint 的存储路径必须可靠不能是临时目录。3.4 分区数对持久化效果的影响持久化效果和分区数密切相关。分区太少每个分区数据量大内存放不下容易溢写分区太多每个分区数据量小但管理开销和序列化开销累加起来也不容忽视。我的经验法则是持久化后的每个分区大小控制在 100MB 到 200MB 之间。假设数据集大小是 10GB那么分区数设在 50 到 100 之间比较合适。如果原始分区数不合适在 persist 之前用repartition()或coalesce()调整。注意 repartition 会触发 shuffle代价不小所以最好在数据量还小的时候就把分区数调好或者用 coalesce 减少分区不触发 shuffle但可能造成数据倾斜。4. 常见问题与排查技巧实录4.1 缓存没生效的几种典型情况情况一persist 之后没有行动操作。这是最常见的。persist 是惰性的必须跟一个行动操作才会触发缓存。检查方法看 Spark UI 的 Storage 页签如果 RDD 没有出现在列表里说明缓存没发生。情况二缓存被驱逐了。MEMORY_ONLY 级别下如果内存不够Spark 会直接丢弃部分分区的缓存不会报错。后续用到这些分区时重新计算。检查方法Storage 页签里看 Fraction Cached如果小于 1.0说明有分区没缓存上。解决办法是换 MEMORY_AND_DISK 级别或者增加内存、减少分区数据量。情况三RDD 被重新创建了。如果你在代码里多次调用sc.textFile()或sc.parallelize()创建同名 RDD它们是不同的对象缓存不共享。确保持久化的 RDD 对象被复用而不是每次重新创建。4.2 内存溢出与 GC 频繁的排查思路持久化用不好反而会加剧内存问题。典型症状是任务跑着跑着报 OOM或者 GC 时间占比超过 20%。排查步骤先看 Spark UI 的 Executor 页签观察每个 Executor 的内存使用曲线。如果内存持续高位、GC 频繁说明缓存占用了太多内存。这时候要么降低存储级别从 MEMORY_ONLY 换成 MEMORY_AND_DISK_SER要么减少缓存的数据量只缓存真正复用的 RDD不要什么都缓存。另一个容易忽略的点是缓存的对象如果包含大量小对象Java 对象头开销会非常可观。比如一个 RDD 里存的是(String, Int)的元组每个元组本身就是一个对象加上 String 对象和 Integer 装箱实际内存占用可能是原始数据的 5 倍以上。这种情况下用 MEMORY_ONLY_SER 配合 Kryo 序列化器效果会好很多。4.3 序列化器选择对持久化的影响Spark 默认用 Java 序列化器速度慢、体积大。换成 Kryo 序列化器体积通常能减少 30% 到 50%速度也更快。配置方法sc SparkContext(local[4], KryoDemo) sc.getConf().set(spark.serializer, org.apache.spark.serializer.KryoSerializer) sc.getConf().set(spark.kryo.registrationRequired, false)注意 Kryo 需要注册自定义类才能发挥最佳效果不注册也能用但会写入类名信息体积略大。对于简单的 String、Int 等类型Kryo 有内置支持不用额外注册。4.4 常见问题速查表问题现象可能原因排查方法解决方案任务变慢日志显示重复计算未持久化或缓存失效看 DAG 中是否有重复的 Stage对复用 RDD 调用 persistStorage 页签无缓存记录persist 后无行动操作检查代码执行顺序在 persist 后加行动操作Fraction Cached 1.0内存不足分区被驱逐看 Executor 内存使用换 MEMORY_AND_DISK 或增内存GC 时间占比高缓存对象太大或太多看 GC 日志和内存曲线换序列化存储级别减少缓存量checkpoint 后任务反而变慢未先 persist重复计算检查 persist 和 checkpoint 顺序先 persist 再 checkpoint缓存 RDD 读取速度慢序列化反序列化开销大对比 MEMORY_ONLY 和 SER 的耗时内存充足时用非序列化级别4.5 几个我踩过的坑第一个坑在循环里反复 persist 同一个 RDD。有一次写迭代算法每一轮迭代都调用了rdd.persist()结果 Storage 页签里出现了几十个缓存条目内存直接被撑爆。正确做法是在循环外 persist 一次循环内复用。第二个坑checkpoint 目录设在了本地文件系统。本地路径在集群模式下每个 Executor 看到的是自己的本地盘checkpoint 数据写散了后续读取时找不到完整数据。必须用 HDFS 或对象存储这类共享文件系统。第三个坑unpersist 之后又用了这个 RDD。unpersist 会释放缓存但 RDD 对象还在后续行动操作会触发重新计算。如果这个 RDD 的血缘链很长重新计算代价很大。所以 unpersist 之前要确认这个 RDD 真的不再用了。5. 不同场景下的持久化策略选型5.1 迭代计算场景机器学习训练、图计算这类迭代场景数据集会被反复扫描几十上百次。策略是在迭代开始前 persist 训练数据存储级别用 MEMORY_AND_DISK_SER 或 MEMORY_ONLY_SER根据内存大小定。如果迭代轮次很多超过 50 轮考虑在中间轮次做一次 checkpoint切断血缘链防止 Driver 端血缘序列化开销过大。5.2 ETL 流水线场景典型的 ETL 流程是读数据、清洗、转换、聚合、写出。如果中间某个清洗后的 RDD 被多个下游分支使用比如既要统计指标又要导出明细那这个 RDD 就值得 persist。存储级别用 MEMORY_AND_DISK因为 ETL 数据量通常较大纯内存扛不住。如果只是单链路流转没有复用那不需要 persist直接流式处理即可。5.3 交互式查询场景用 Spark SQL 或 Dataset API 做交互式查询时经常需要对同一份数据跑多个查询。这时候可以把基础表 cache 起来。Spark SQL 里可以用spark.catalog.cacheTable(tableName)或dataFrame.cache()。存储级别默认是 MEMORY_AND_DISK可以通过spark.sql.inMemoryColumnarStorage.compressed开启压缩进一步节省空间。5.4 流处理场景Structured Streaming 里持久化和批处理不太一样。流处理的数据是持续到来的缓存的是流式 DataFrame 的中间状态。通常用writeStream的foreachBatch里做缓存或者用spark.sql.streaming.stateStore相关的配置管理状态。这块比较复杂建议单独查官方文档不要照搬批处理的持久化策略。6. 持久化与 Spark 内存管理模型的配合6.1 统一内存管理下的缓存空间Spark 1.6 之后引入了统一内存管理模型Execution 内存和 Storage 内存共享一个池子。默认情况下Storage 内存占统一内存的 50%但 Execution 可以借用 Storage 的空闲空间反过来 Storage 也可以借用 Execution 的空闲空间。这意味着缓存不是固定占那么多内存而是动态调整的。这个机制的好处是灵活坏处是不确定性。你 persist 了一个 RDD以为它会一直缓存着结果某个 Stage 的 shuffle 需要大量 Execution 内存把 Storage 的内存借走了缓存被驱逐。所以对于关键 RDD要么用 MEMORY_AND_DISK 级别保证溢写磁盘不丢数据要么调大spark.memory.storageFraction参数给 Storage 更多保障。6.2 堆外内存的使用OFF_HEAP 存储级别把数据存在堆外内存不受 JVM GC 管理。好处是减少 GC 压力坏处是需要配置spark.memory.offHeap.enabled和spark.memory.offHeap.size而且堆外内存的读写速度略慢于堆内。适合内存充足、GC 问题严重的场景。我一般不建议新手一上来就用 OFF_HEAP先把堆内内存用好再说。6.3 动态资源分配对缓存的影响开启动态资源分配spark.dynamicAllocation.enabledtrue后Executor 会根据负载动态增减。如果某个 Executor 被回收了它上面的缓存数据也就丢了。虽然后续可以重算但如果血缘链很长重算代价很大。所以开启动态资源分配时要么用 MEMORY_AND_DISK 级别把数据溢写到磁盘要么设置spark.dynamicAllocation.minExecutors保证最少有足够的 Executor 持有缓存。7. 性能对比实测与调优建议7.1 不同存储级别的耗时对比我在一个模拟项目里做过一组对比测试数据集约 5GB集群 4 个 Executor每个 8GB 内存。测试逻辑是对同一个 RDD 连续做 5 次 count 操作对比不同存储级别的总耗时存储级别第一次耗时后续每次耗时总耗时内存占用无持久化45s45s225s低MEMORY_ONLY48s2s56s高接近上限MEMORY_ONLY_SER52s4s68s中等MEMORY_AND_DISK_SER55s6s79s低DISK_ONLY60s15s120s极低从数据看MEMORY_ONLY 总耗时最短但内存占用最高。MEMORY_ONLY_SER 总耗时略长但内存占用降了一半多。MEMORY_AND_DISK_SER 在内存有限的情况下是最稳妥的选择。DISK_ONLY 虽然内存占用极低但每次读取都要走磁盘 IO耗时明显增加。7.2 调优建议清单优先 persist 那些被多次使用的 RDD只使用一次的 RDD 不要 persist。内存充足用 MEMORY_ONLY内存中等用 MEMORY_ONLY_SER内存紧张用 MEMORY_AND_DISK_SER。序列化器换成 Kryo体积和速度都有明显改善。持久化前调整分区数每个分区 100MB 到 200MB 为宜。迭代计算超过 50 轮时中间做一次 checkpoint。用完的 RDD 及时 unpersist释放内存。开启动态资源分配时用磁盘溢写级别防止缓存丢失。监控 Spark UI 的 Storage 页签关注 Fraction Cached 和内存使用曲线。7.3 一个完整的调优案例某次处理一个 20GB 的日志分析任务原始代码没有做任何持久化跑一次要 25 分钟。分析 DAG 后发现清洗后的 RDD 被三个下游分支使用一个统计 PV、一个统计 UV、一个导出明细。三个分支各自触发了一次完整的清洗计算。优化方案在清洗后的 RDD 上调用persist(StorageLevel.MEMORY_AND_DISK_SER)分区数从默认的 200 调整为 100每个分区约 200MB。同时把序列化器换成 Kryo。优化后跑一次耗时降到 8 分钟其中第一次计算加缓存用了 5 分钟后续两个分支各 1.5 分钟。内存使用稳定在 60% 左右没有出现 OOM 或频繁 GC。这个案例说明持久化的收益在复用场景下非常显著。但前提是你要先看清楚 DAG找到真正被复用的 RDD而不是盲目地给所有 RDD 加 persist。8. 持久化策略的边界与注意事项持久化不是银弹有些场景下它反而会拖慢任务。比如单次使用的 RDDpersist 之后白白占内存还多了一次写缓存的开销。再比如数据量极小、计算极快的 RDD缓存的开销可能比重新计算还大。还有一个容易被忽略的点persist 的 RDD 如果包含大量分区每个分区的缓存元数据也会占用 Driver 内存。分区数上万的时候Driver 端的管理开销不容忽视。这种情况下要么减少分区数要么用 checkpoint 替代 persist。最后提醒一点持久化策略不是一劳永逸的。数据量变了、集群规模变了、业务逻辑变了最优的存储级别和分区数都可能需要重新调整。我习惯在每次大的数据量波动后重新跑一次基准测试确认当前的持久化配置仍然合理。这个习惯帮我避免了好几次潜在的性能退化。