1. 为什么你的Spark作业跑得慢从一次真实的调优经历说起刚接触Spark那会儿我写过一个看起来特别简单的作业读一份日志数据做几轮过滤和聚合最后把结果写到文件系统。逻辑上没有任何问题代码也就二十来行但跑起来慢得让人抓狂。更诡异的是我盯着任务监控界面看发现同一个中间结果被反复计算了好几遍——明明上一步已经算出来了下一步又从头算一遍像是一个记性特别差的人每做一道菜都要重新去菜市场买一次菜。后来我才明白问题出在RDD的惰性求值机制上。Spark的RDD默认不会保存中间计算结果每次遇到行动操作Action它都会沿着血缘关系Lineage从头重新计算一遍。这在单次计算场景下没问题但一旦你的作业里有多个行动操作或者某个RDD被多个下游分支复用重复计算的开销就会成倍放大。解决这个问题的核心手段就是RDD持久化Persistence也叫缓存Caching。这篇文章就是围绕RDD持久化展开的。我会从底层原理讲到实操细节把存储级别的选择逻辑、缓存与检查点的区别、序列化的取舍、内存管理的影响这些容易踩坑的地方全部拆开讲清楚。不管你是刚学Spark的新手还是已经写过一些作业但总觉得性能差口气的开发者这篇内容都能帮你把RDD持久化这件事彻底搞明白。关键词就三个RDD持久化、缓存策略、存储级别全文围绕它们展开。2. RDD惰性求值带来的重复计算问题持久化到底在解决什么2.1 血缘关系与惰性求值Spark的设计哲学要理解持久化得先理解Spark为什么会有重复计算这个问题。RDD弹性分布式数据集是Spark最核心的数据抽象它有两个关键特性不可变和惰性求值。不可变意味着你不能修改一个已有的RDD只能通过转换操作Transformation生成新的RDD。比如你有一个RDD叫rdd1对它做map操作得到rdd2rdd1本身不会变rdd2是一个全新的RDD。每个新RDD都会记录自己是从哪个父RDD、经过什么操作得来的这条链路就是血缘关系Lineage。惰性求值则意味着当你调用map、filter、flatMap这类转换操作时Spark并不会立刻执行计算它只是把这条转换记录到血缘关系里。真正触发计算的是行动操作Action比如count、collect、saveAsTextFile、foreach等。只有遇到行动操作Spark才会根据血缘关系从源头开始一步步把计算跑完。这个设计的好处是Spark可以在执行前做整体优化比如合并窄依赖、调整分区等。但代价就是每次行动操作都会触发一次完整的血缘回溯计算。2.2 重复计算是怎么发生的一个具体的场景假设你写了这样一段代码rdd1 sc.textFile(hdfs://path/to/logs) rdd2 rdd1.filter(lambda line: ERROR in line) rdd3 rdd2.map(lambda line: extract_fields(line)) count rdd3.count() # 第一个行动操作 rdd3.saveAsTextFile(output) # 第二个行动操作这里rdd3被用了两次一次是count()一次是saveAsTextFile()。由于RDD默认不保存中间结果这两次行动操作会各自触发一次从rdd1到rdd3的完整计算。也就是说读文件、过滤、字段提取这一整套流程会被执行两遍。如果rdd1的数据量很大或者extract_fields这个函数计算很复杂那第二遍计算就是纯粹的资源浪费。更糟糕的情况是迭代式计算比如机器学习算法里常见的循环data sc.textFile(hdfs://path/to/data).cache() for i in range(100): data data.map(update_function) result data.reduceByKey(aggregate) print(result.count())如果data没有做持久化每一轮循环都会从最初的textFile开始重新计算100轮就是100次全量重算。这在生产环境里是不可接受的。2.3 持久化的本质用空间换时间RDD持久化的思路非常直接把计算出来的RDD数据存到内存或磁盘里下次需要用到这个RDD时直接从存储介质里读取而不是重新计算。用生活化的类比来说这就像你做菜时把切好的食材先装盘放冰箱。第一次切菜花了十分钟但后面每次要用这些食材时直接从冰箱拿出来就行不用重新切一遍。代价是你得占用冰箱空间而且食材放久了可能不新鲜对应内存不足时数据被淘汰的情况。在Spark里持久化通过persist()或cache()方法实现。cache()是persist()的一个特例等价于persist(StorageLevel.MEMORY_ONLY)。调用这两个方法后RDD并不会立刻被缓存而是打上一个标记。等到第一次遇到行动操作、真正计算出这个RDD的数据时Spark才会把数据按照指定的存储级别保存下来。后续再遇到需要这个RDD的行动操作就直接从缓存读取。这里有一个关键点需要强调持久化是懒执行的。你调用persist()的时候什么都不会发生只有第一次行动操作触发计算后缓存才会真正建立。这个特性经常让新手困惑——明明写了cache()为什么感觉没生效原因往往就是没有触发行动操作或者缓存的数据已经被淘汰了。3. 存储级别怎么选从MEMORY_ONLY到OFF_HEAP的完整对比3.1 StorageLevel的五个维度Spark的StorageLevel类定义了持久化的具体方式它由五个布尔参数组合而成参数含义useDisk是否使用磁盘useMemory是否使用内存useOffHeap是否使用堆外内存deserialized是否以反序列化形式存储replication副本数量通常为1或2这五个参数的组合构成了Spark预定义的十几种存储级别。理解每个参数的含义是选对存储级别的前提。useMemory和useDisk决定了数据存在哪里。内存读写速度快但容量有限磁盘容量大但读写慢。两者可以同时为true表示优先放内存内存不够时溢写到磁盘。deserialized决定了数据以什么形式存储。反序列化形式即Java对象形式读取快因为不需要额外的反序列化开销但占用内存大通常是序列化形式的2到5倍。序列化形式如字节数组占用空间小但每次读取都需要反序列化消耗CPU。useOffHeap表示使用堆外内存通过Tungsten项目的内存管理机制。堆外内存不受JVM垃圾回收的直接影响可以减少GC压力但需要额外的配置和管理。replication是副本数默认为1。设为2时每个分区的数据会在两个节点上各存一份提高容错性但占用双倍存储空间。3.2 常用存储级别的适用场景Spark预定义的存储级别有十几种但实际工作中常用的就那么几个。我把它们整理成一张表方便你对照选择存储级别内存磁盘序列化副本适用场景MEMORY_ONLY是否否1数据量小、内存充足、需要频繁读取MEMORY_AND_DISK是是否1数据量中等、内存可能不够、通用场景MEMORY_ONLY_SER是否是1数据量较大、内存紧张、能接受反序列化开销MEMORY_AND_DISK_SER是是是1数据量大、内存有限、生产环境常用DISK_ONLY否是是1数据量极大、内存严重不足MEMORY_AND_DISK_2是是否2对容错性要求高、可接受双倍存储OFF_HEAP否否是1需要减少GC影响、配置了堆外内存MEMORY_ONLY是最简单的级别也是cache()的默认行为。它把RDD的每个分区以反序列化的Java对象形式存在内存里。如果某个分区的数据放不下这个分区就不会被缓存下次用到时重新计算。注意是整个分区不缓存而不是部分缓存。所以当内存不足时你可能会发现有些分区命中了缓存有些没有导致性能表现不稳定。MEMORY_AND_DISK是我个人最推荐的通用级别。它优先把数据放内存内存放不下的分区溢写到磁盘。这样至少保证了数据不会丢失虽然磁盘读取比内存慢但比重新计算整条血缘链路要快得多。对于大多数中等规模的数据处理作业这个级别是稳妥的选择。MEMORY_ONLY_SER和MEMORY_AND_DISK_SER的区别在于序列化。序列化后数据占用空间更小同样大小的内存能缓存更多数据。代价是每次读取都要反序列化消耗CPU。如果你的数据量大、内存紧张但CPU资源相对充裕选SER版本是划算的。DISK_ONLY适合数据量极大、内存完全放不下的场景。虽然慢但至少避免了重复计算。不过在实际生产中如果数据大到只能放磁盘通常意味着你需要重新考虑分区策略或者作业设计而不是单纯依赖DISK_ONLY。OFF_HEAP需要额外配置spark.memory.offHeap.enabled和spark.memory.offHeap.size。它的优势是绕过JVM堆内存管理减少GC停顿。但配置不当容易出问题建议在对GC性能有明确需求时再使用。3.3 选型决策的实操逻辑面对这么多级别怎么选我总结了一个简单的决策流程第一步看数据量。如果数据量明显小于可用内存比如只占30%以下直接用MEMORY_ONLY简单高效。如果数据量接近或超过内存进入第二步。第二步看CPU和内存的权衡。如果内存紧张但CPU核数多选MEMORY_ONLY_SER用CPU换内存。如果两者都紧张选MEMORY_AND_DISK_SER。第三步看容错要求。如果集群节点不稳定或者作业不能容忍数据丢失考虑加副本_2后缀。但副本会占用双倍空间要确保存储资源足够。第四步看GC表现。如果监控发现GC停顿频繁且时间长可以尝试OFF_HEAP但要做好配置调优。提示存储级别的选择不是一劳永逸的。同一个作业在不同数据规模、不同集群配置下最优级别可能不同。建议在开发阶段多做对比测试用Spark UI观察缓存命中率和GC情况。4. cache与persist的差异、checkpoint的定位以及它们之间的配合4.1 cache和persist看似相同实则不同很多初学者会把cache()和persist()当成一回事因为它们确实很像。cache()内部就是调用了persist(StorageLevel.MEMORY_ONLY)。但它们的区别在于灵活性和语义。cache()是一个快捷方法只支持默认的MEMORY_ONLY级别。如果你需要其他存储级别必须用persist()。所以从功能上讲persist()是cache()的超集。但从使用语义上我更倾向于这样区分cache()适合那些数据量不大、内存肯定放得下、我就想简单缓存一下的场景persist()适合需要精细控制存储策略的场景。在实际项目中我几乎都用persist()因为显式指定存储级别能让代码意图更清晰也方便后续调优。还有一个细节cache()和persist()都是懒执行的而且只能对还没有被计算过的RDD调用。如果你先对RDD做了行动操作再调用persist()那这个RDD的数据已经计算过了但不会被缓存你需要重新触发一次行动操作才能建立缓存。这个坑我踩过当时排查了半天才发现是调用顺序的问题。4.2 checkpoint切断血缘关系的重型武器checkpoint检查点是另一个和持久化相关的机制但它的定位和persist完全不同。persist是把数据缓存起来但血缘关系依然保留。如果缓存的数据丢失了比如节点故障导致内存数据丢失Spark还可以根据血缘关系重新计算。血缘关系就像一份菜谱即使冰箱里的食材坏了你还能照着菜谱重新做。checkpoint则是把RDD的数据写入可靠的分布式文件系统如HDFS并且切断血缘关系。一旦做了checkpoint这个RDD就不再依赖父RDD了它自己成了一个新的数据源。这样做的好处是对于血缘关系特别长的RDD比如迭代1000次的机器学习模型可以避免血缘链路过长导致的恢复开销和栈溢出风险。checkpoint和persist的另一个关键区别是执行时机。persist是在第一次行动操作时缓存数据而checkpoint会额外触发一个专门的作业来完成数据写入。这意味着checkpoint的开销更大不能随便用。4.3 为什么推荐persist和checkpoint配合使用在实际项目中我通常会把persist和checkpoint配合使用尤其是在迭代计算场景下。原因在于checkpoint会触发一个独立的作业来写数据如果这个RDD没有被持久化那这个独立作业会从头计算一遍血缘链路开销很大。而如果先persistcheckpoint作业就可以直接从缓存读取数据写入文件系统省去重复计算。推荐的写法是这样的rdd sc.textFile(hdfs://path/to/data) rdd rdd.map(complex_transformation).filter(condition) # 先持久化再设置检查点 rdd.persist(StorageLevel.MEMORY_AND_DISK_SER) rdd.checkpoint() # 触发行动操作此时会先建立缓存再执行checkpoint rdd.count()注意checkpoint()调用后也需要一个行动操作来触发。而且checkpoint的目录需要通过sc.setCheckpointDir()提前设置好。注意checkpoint会切断血缘关系这意味着一旦checkpoint完成你就不能再通过这个RDD回溯到它的父RDD了。如果后续逻辑需要用到父RDD要提前做好规划。5. 缓存失效、内存淘汰与序列化的那些坑5.1 缓存什么时候会失效RDD缓存建立后并不是永久有效的。以下几种情况会导致缓存失效或数据丢失内存不足导致的淘汰。Spark的内存管理采用LRU最近最少使用策略。当内存空间不够时Spark会淘汰最久未使用的缓存分区。被淘汰的分区下次用到时需要重新计算。这就是为什么MEMORY_ONLY级别下你可能会发现缓存命中率不稳定——内存压力大的时候刚缓存的数据可能很快就被挤出去了。Executor故障。如果某个Executor节点挂了它上面缓存的所有RDD分区都会丢失。Spark会调度这些分区到其他节点重新计算。这也是为什么对容错性要求高的场景要考虑加副本。手动调用unpersist。rdd.unpersist()会显式释放缓存。这个操作在RDD不再需要时很有用可以及时释放内存给其他RDD使用。但要注意unpersist是阻塞的它会等所有分区的缓存都释放完才返回。作业结束。SparkContext关闭后所有缓存自然失效。5.2 序列化带来的性能权衡选择SER级别的存储时序列化框架的选择会影响性能。Spark默认使用Java序列化但它比较慢序列化后的数据也偏大。在生产环境中我强烈建议使用Kryo序列化。Kryo的配置方式是在SparkConf里设置conf SparkConf() conf.set(spark.serializer, org.apache.spark.serializer.KryoSerializer) conf.set(spark.kryo.registrationRequired, false)如果设为true你需要显式注册所有要序列化的类这样性能更好但配置更麻烦。设为false时Kryo会自动处理未注册的类但会输出警告信息。Kryo相比Java序列化的优势很明显序列化速度快、体积小。实测下来同样的数据用Kryo序列化后体积通常能减少30%到50%反序列化速度也快不少。对于需要频繁读写缓存的迭代作业这个提升很可观。但Kryo也有坑。它不支持所有Java类型某些自定义类可能需要编写自定义序列化器。另外Kryo序列化后的数据在不同Spark版本之间可能不兼容升级Spark版本时要重新生成缓存。5.3 缓存位置对性能的实际影响数据缓存在内存的哪个位置对性能也有影响。Spark的内存分为执行内存Execution Memory和存储内存Storage Memory两块。执行内存用于shuffle、join、sort等操作存储内存用于缓存RDD。在Spark 1.6之后这两块内存可以动态借用。默认情况下存储内存占统一内存的50%左右具体比例由spark.memory.fraction和spark.memory.storageFraction控制。如果执行内存不够它会向存储内存借这可能导致缓存数据被挤掉。如果你的作业既有大量shuffle操作又需要缓存重要数据可能需要调整spark.memory.storageFraction给存储内存留更多空间。但这样会压缩执行内存可能拖慢shuffle。这是一个需要根据实际负载权衡的参数。提示在Spark UI的Storage标签页里可以看到每个RDD的缓存情况包括缓存了多少分区、占用了多少内存、有多少被淘汰。这是排查缓存问题的第一手资料。6. 从开发到生产持久化策略的落地经验6.1 什么时候该持久化什么时候不该持久化不是越多越好。每个被缓存的RDD都会占用内存如果缓存了太多RDD内存被占满反而会导致频繁的淘汰和重算性能可能比不缓存还差。我的经验是只对满足以下条件的RDD做持久化被多个行动操作复用。如果一个RDD只被一个行动操作使用缓存它没有意义因为计算一次就结束了。计算代价高。如果RDD的血缘链路很长或者中间有复杂的转换如正则匹配、复杂UDF缓存能省下大量重复计算。数据量适中。数据量太小缓存收益不明显数据量太大缓存放不下反而添乱。迭代计算中的基础数据。机器学习、图计算等迭代场景每轮循环都依赖的基础RDD必须持久化。反过来以下情况不建议持久化只被使用一次的RDD、计算非常简单的RDD如简单的map、数据量远超内存且没有磁盘溢写需求的RDD。6.2 一个完整的调优案例回到开头提到的那个慢作业。原始代码是这样的rdd1 sc.textFile(hdfs://path/to/logs) rdd2 rdd1.filter(lambda line: ERROR in line) rdd3 rdd2.map(lambda line: extract_fields(line)) count rdd3.count() rdd3.saveAsTextFile(output)问题很明显rdd3被用了两次但没有持久化导致整条链路计算了两遍。优化后的代码rdd1 sc.textFile(hdfs://path/to/logs) rdd2 rdd1.filter(lambda line: ERROR in line) rdd3 rdd2.map(lambda line: extract_fields(line)) # 持久化rdd3选择MEMORY_AND_DISK_SER rdd3.persist(StorageLevel.MEMORY_AND_DISK_SER) count rdd3.count() # 第一次触发建立缓存 rdd3.saveAsTextFile(output) # 第二次直接从缓存读取 # 用完后释放缓存 rdd3.unpersist()为什么选MEMORY_AND_DISK_SER而不是MEMORY_ONLY因为日志数据量通常不小MEMORY_ONLY可能放不下导致部分分区不缓存。MEMORY_AND_DISK_SER保证了所有分区都能缓存内存放不下就写磁盘序列化又减少了内存占用。实测下来这个作业的执行时间从原来的十几分钟降到了三分钟左右。6.3 监控与验证怎么确认缓存真的生效了写完持久化代码后怎么确认它真的起作用了我通常从三个地方验证Spark UI的Storage标签页。这里会列出所有被持久化的RDD显示缓存的分区数、内存占用、磁盘占用。如果某个RDD显示Fraction Cached是100%说明全部缓存成功如果是50%说明有一半分区没缓存上需要调整存储级别或内存配置。Spark UI的Stages标签页。观察每个Stage的输入数据量。如果第二次使用同一个RDD时输入数据量明显小于第一次说明缓存命中了。如果两次输入量一样说明缓存没生效。日志中的缓存相关信息。Spark会在日志里输出缓存的建立和淘汰信息。开启DEBUG级别日志可以看到更详细的内容但生产环境一般用INFO级别就够了。6.4 几个容易忽略的细节最后分享几个我在实际项目中踩过的坑和总结的经验第一持久化要在行动操作之前调用。这个前面提过但真的很重要。我见过不少代码把persist()写在count()后面然后疑惑为什么没缓存。第二注意RDD的引用。如果你对RDD做了持久化但后续代码里又对这个RDD做了转换生成新RDD那新RDD不会自动继承持久化属性。比如rdd.persist()之后又写了rdd2 rdd.map(...)rdd2是没有缓存的只有rdd有。第三及时释放不再需要的缓存。在长时间运行的作业里如果不断创建新的持久化RDD而不释放旧的内存会被逐渐占满。养成在RDD不再使用时调用unpersist()的习惯。第四不要过度依赖缓存来解决性能问题。缓存只是优化手段之一如果作业本身的分区策略不合理、数据倾斜严重、shuffle量过大光靠缓存解决不了根本问题。先排查这些基础问题再考虑缓存优化。第五测试环境验证后再上生产。存储级别的选择、内存参数的调整都建议先在测试环境用小规模数据验证观察缓存命中率和GC情况确认有效后再应用到生产环境。直接在生产环境调参风险很大。