Spark 2.2.0（EMR环境）缓存分区DataFrame后分区失效求助

我之前在处理类似的大分区DataFrame缓存问题时也踩过这个坑，Spark默认的缓存机制确实会让你丢失分区裁剪的优势——原因很简单：当你调用cache()或者默认的persist()时，Spark会把整个DataFrame的数据加载到内存（或磁盘），但缓存后的数据集会失去原始的分区索引优化，导致查询时不管你加了什么分区过滤条件，都会全量扫描缓存里的数据。

针对你的Spark 2.2.0场景，这里有几个靠谱的解决方案，能帮你保留分区特性同时实现高效缓存：

1. 使用分区感知的持久化策略 #

不要用默认的cache()，而是显式指定persist的存储级别为序列化存储（比如MEMORY_AND_DISK_SER），这种方式会保留DataFrame的原始分区结构，同时让Spark优化器能识别分区过滤条件：

import org.apache.spark.storage.StorageLevel

// 显式指定序列化存储级别，保留分区结构
val cachedDf = df.persist(StorageLevel.MEMORY_AND_DISK_SER)

序列化存储不仅能减少内存占用（比反序列化对象省空间），更重要的是能让Spark在查询时依然执行分区裁剪，只扫描符合条件的分区数据。

2. 缓存分区RDD再转回DataFrame #

RDD的分区特性比DataFrame更“硬核”，你可以先把DataFrame转成RDD缓存（保留原始分区），再转回DataFrame，这样查询时的分区裁剪逻辑会完全生效：

import org.apache.spark.storage.StorageLevel

// 转换为RDD并缓存，保留原始分区
val partitionedRdd = df.rdd.persist(StorageLevel.MEMORY_AND_DISK_SER)
// 转回DataFrame，保持分区结构
val cachedDf = spark.createDataFrame(partitionedRdd, df.schema)

这种方式在Spark 2.x版本中非常稳定，因为RDD的分区信息不会被缓存操作破坏，查询cachedDf.filter($"k1" === v1)时只会扫描对应k1的分区。

3. 注册分区临时表并缓存 #

这是最推荐的方案，因为分区表本身就带有分区索引，缓存时会直接保留这个索引结构：

// 先将DataFrame保存为分区临时表
df.write.mode("overwrite")
  .partitionBy("k1", "k2")
  .saveAsTable("temp_partitioned_df")

// 缓存这个分区表
spark.sql("CACHE TABLE temp_partitioned_df")

// 查询时直接使用分区过滤，只会扫描对应分区
val result = spark.sql("SELECT * FROM temp_partitioned_df WHERE k1 = 'v1'")

临时表的分区信息会被Spark的Catalog记录，缓存时会针对每个分区单独缓存，查询时自然会触发分区裁剪，完全不会出现全量扫描的问题。

4. 按需缓存特定分区（最省资源） #

如果你的查询总是集中在特定的k1或k2值上，那完全没必要缓存整个40G的DataFrame，只缓存你需要的分区即可：

// 只缓存k1=v1的分区，内存占用仅为原DataFrame的2%
val targetDf = df.filter($"k1" === "v1").persist(StorageLevel.MEMORY_AND_DISK_SER)

这种方式不仅能避免内存溢出，查询速度还会比全量缓存快很多，因为数据量小了一个数量级。

补充说明 #

为什么默认缓存会失效？因为Spark默认的MEMORY_ONLY存储级别会把数据以反序列化对象的形式存在内存，缓存后的DataFrame会被视为一个“扁平”的数据集，优化器会跳过分区裁剪逻辑，直接扫描所有缓存数据——这对40G的数据集来说自然会导致OOM或性能暴跌。

内容的提问来源于stack exchange，提问作者rongenre