PySpark显式缓存DataFrame后底层文件更新触发java.io.FileNotFoundException问题问询

你遇到的缓存后仍触发源文件读取报错的问题，是Spark的缓存机制和DAG重计算逻辑的固有特性导致的，核心原因有两点：

• cache()默认使用MEMORY_ONLY存储级别，只要出现部分缓存分区因内存不足被清理、对应Executor宕机缓存丢失的情况，Spark会自动回溯血缘重新计算这部分分区，此时就会回源读取文件，刚好源文件被外部修改/删除就会抛出java.io.FileNotFoundException。
• 你用show(1)触发缓存只会计算show操作需要的少量分区，全量分区并没有全部完成缓存，剩下未被触发计算的分区第一次被使用时，仍然会直接读取源文件，此时源文件变更也会触发报错。

缓存的核心作用是减少重复计算的开销，只要缓存的分区有效就不会回源读取，但它的设计定位本身不包含"冻结"源数据、100%避免重算的能力，所以出现上述场景属于正常逻辑。

方案1：轻量改代码，全量固化缓存 #

这是性能最高、改动最小的方案，只要你的集群内存+磁盘空间足够存储该表数据，全程不会再回源读取：

from pyspark import StorageLevel

# 改用MEMORY_AND_DISK存储级别，内存不足时自动将缓存落盘，不会丢失缓存分区
a.user_profile_df = user_profile_df.persist(StorageLevel.MEMORY_AND_DISK)
# 触发全量分区计算缓存，不要用show(1)，count()会扫描所有分区确保全部写入缓存
a.user_profile_df.count()

方案2：大数据量场景临时固化文件 #

如果表数据量过大无法全量缓存，该方案比你当前用CSV存储的性能高3~10倍：

# 将需要的分区写入临时Parquet（列式存储，压缩比和读写性能远高于CSV）
user_profile_df.write.parquet("hdfs://你的自定义临时目录/user_profile_temp.parquet", mode="overwrite")
# 后续全流程都读取这份临时文件，不会再碰原表的源文件
a.user_profile_df = spark.read.parquet("hdfs://你的自定义临时目录/user_profile_temp.parquet")

方案3：调度协同规避冲突 #

如果你们的任务调度时间可控，可以调整你的脚本运行时间，避开该表的更新时间窗口；如果需要读取最新的表数据，可以在读取表之前执行spark.sql("REFRESH TABLE user_table")刷新元数据，确保读取到最新的文件路径，不会访问已经被删除的旧文件。

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