Spark缓存

在 Spark 中，缓存（Caching）是一种优化技术，用于将中间计算结果存储在内存或磁盘中，避免重复计算，从而显著提升迭代计算或交互式查询的性能。以下是关于 Spark 缓存的详细介绍：

一、为什么需要缓存？

Spark 的 RDD 操作是惰性的，每次触发行动算子（如collect()、count()）时都会重新计算整个 DAG（有向无环图）。对于需要多次使用的 RDD（如迭代算法或交互式查询），这种重复计算会造成极大的资源浪费。缓存可以将中间结果持久化，避免重复计算。

二、缓存的基本用法

Spark 提供了两种缓存方法：

cache()：默认将 RDD 存储在内存中（等同于persist(StorageLevel.MEMORY_ONLY)）。
persist(StorageLevel)：可指定存储级别（如内存、磁盘、序列化等）。

示例：缓存 RDD

python

运行

from pyspark import SparkContextsc = SparkContext("local", "CacheExample")# 创建RDD
rdd = sc.textFile("hdfs://path/to/large/file.txt")# 缓存RDD
rdd.cache()  # 等同于 rdd.persist(StorageLevel.MEMORY_ONLY)# 第一次行动操作：触发计算并缓存结果
count1 = rdd.count()  # 第一次计算，耗时较长# 第二次行动操作：直接使用缓存结果
count2 = rdd.count()  # 直接从缓存读取，耗时极短

三、存储级别（StorageLevel）

Spark 支持多种存储级别，可根据数据规模和内存情况选择：

存储级别	说明
`MEMORY_ONLY`	默认级别，将 RDD 作为反序列化的 Java 对象存储在内存中。内存不足时部分数据会被丢弃。
`MEMORY_ONLY_SER`	将 RDD 作为序列化的 Java 对象存储（占用空间更小）。
`MEMORY_AND_DISK`	优先存储在内存中，内存不足时溢写到磁盘。
`MEMORY_AND_DISK_SER`	类似`MEMORY_AND_DISK`，但数据序列化存储。
`DISK_ONLY`	只存储在磁盘上。
`MEMORY_ONLY_2`	类似`MEMORY_ONLY`，但数据复制到两个节点。
`OFF_HEAP`	存储在 Tungsten 堆外内存中（需要启用堆外内存）。

示例：指定存储级别

python

运行

from pyspark import StorageLevel# 存储在内存和磁盘，序列化
rdd.persist(StorageLevel.MEMORY_AND_DISK_SER)# 只存储在磁盘
rdd.persist(StorageLevel.DISK_ONLY)

四、缓存的生命周期

缓存何时触发？
首次触发行动算子时，Spark 会计算 RDD 并将结果缓存。
缓存何时失效？
- 手动调用unpersist()释放缓存。
- 内存不足时，Spark 会根据 LRU（最近最少使用）策略自动删除部分缓存数据。
- Spark 应用结束时，缓存自动释放。

示例：释放缓存

python

运行

rdd.unpersist()  # 释放缓存

五、缓存的最佳实践

优先缓存频繁使用的 RDD
如迭代算法中的中间结果（如 MLlib 模型训练）或交互式查询中的数据集。
选择合适的存储级别
- 内存充足时，使用MEMORY_ONLY（性能最优）。
- 数据量大时，使用MEMORY_ONLY_SER或MEMORY_AND_DISK_SER（节省内存）。
- 对容错要求高的场景，使用带副本的存储级别（如MEMORY_ONLY_2）。
避免过度缓存
缓存过多数据会导致内存压力，触发频繁的 GC 或数据溢写磁盘，反而降低性能。
缓存后进行重分区
缓存后可使用coalesce()减少分区数，降低后续任务的调度开销：

python

运行
```
rdd_cached = rdd.cache()
rdd_optimized = rdd_cached.coalesce(10)  # 合并为10个分区
```
监控缓存使用情况
通过 Spark UI 查看缓存状态：
- Storage 标签页显示各 RDD 的缓存大小、分区数和存储位置。
- 监控内存使用，避免 OOM（内存溢出）。

六、缓存 vs 检查点（Checkpointing）

特性	缓存（Cache）	检查点（Checkpoint）
存储位置	内存或磁盘（Executor 节点）	HDFS 等可靠存储（外部系统）
容错性	节点故障时可能丢失数据，需重新计算	数据永久存储，节点故障不影响
性能	速度快（内存读取）	速度较慢（需写入外部存储）
用途	短期重用中间结果	长期保存关键结果（如长时间迭代的中间点）
触发方式	首次行动算子自动触发	手动调用`checkpoint()`并触发行动算子

示例：使用检查点

python

运行

# 设置检查点目录
sc.setCheckpointDir("hdfs://path/to/checkpoint_dir")# 标记RDD为检查点
rdd.checkpoint()# 触发检查点（必须有行动算子）
rdd.count()  # 此时RDD会被写入检查点目录

七、常见问题与解决方法

内存不足导致缓存失效
- 解决方案：改用MEMORY_AND_DISK_SER存储级别，或增加内存资源。
缓存数据丢失
- 解决方案：使用MEMORY_ONLY_2存储级别（数据复制到两个节点），或结合检查点。
缓存未生效
- 检查是否在行动算子前调用了cache()或persist()。
- 确认 RDD 是否被重复创建（每次map()、filter()等转换操作都会生成新 RDD）。