PySpark循环中缓存DataFrame重复读取问题解决方案

缓存未生效、循环重复读取源数据的核心诱因是错误的Spark配置，叠加代码语法问题导致缓存预热失效：

• 你手动设置了spark.storage.memoryFraction=0，该参数控制Executor用于缓存DataFrame/RDD的内存占比，设为0等于直接关闭了内存缓存能力，调用cache()时数据没有存储空间，每次触发action只能从头读取源parquet文件重新计算。
• Python不支持//作为单行注释符，你写在take(1)后的//calling action item...会被识别为代码直接抛出语法错误，原本用于触发缓存预热的action根本没有正常执行，缓存从未被真正初始化。
• 循环内反复union生成的新DataFrame没有做持久化，计算血缘会持续拉长，后续触发action时回溯上游计算链路也可能放大重复读取的问题。

按以下顺序调整代码即可解决重复读取问题：

1. 删除错误的缓存内存配置
   直接删除.config("spark.storage.memoryFraction", 0)这行配置，Spark 2.0+版本已经自动管理存储内存占比，不需要手动设置；如果使用低版本Spark，将该值调整为0.3~0.6区间，为缓存预留足够内存。
2. 修正语法问题，确保缓存正常初始化
   将//注释替换为Python标准的#注释，同时用count()替代take(1)做缓存预热，确保全量分区完成缓存：

def generate_data(spark, logger, conf):
    processed_data_final = None
    path_1 = conf["raw_data_path_1"]
    path_2 = conf["raw_data_path_2"]
    df_path1 = spark.read.parquet(path_1)
    df_path1.cache()
    df_path1.count() # 触发action完成缓存，Spark懒执行机制下不触发action不会真正执行计算和缓存
    df_path2 = spark.read.parquet(path_2)
    df_path2.cache()
    df_path2.count()

3. 优化循环内Union逻辑，避免血缘过长
   循环中每次Union生成的中间结果做缓存，同时及时释放上一版中间结果的缓存，避免血缘无限拉长、内存冗余占用：

for dt in date_list:
        processed_data = process_data(spark, logger, conf, dt, df_path1, df_path2)
        if processed_data_final is None:
            processed_data_final = processed_data
        else:
            last_result = processed_data_final
            processed_data_final = processed_data_final.union(processed_data).cache()
            processed_data_final.count() # 触发action固化当前中间结果
            last_result.unpersist() # 释放上一版中间结果的缓存

    return processed_data_final

4. （可选）调整缓存级别适配资源情况
   如果集群内存资源紧张，可以显式指定序列化缓存级别，降低缓存内存占用：

from pyspark import StorageLevel
# 替换原有的cache()调用
df_path1.persist(StorageLevel.MEMORY_AND_DISK_SER)
df_path2.persist(StorageLevel.MEMORY_AND_DISK_SER)

任务运行时打开Spark UI的Storage页签，能看到df_path1、df_path2对应的缓存条目，且缓存命中率持续上升，即代表缓存生效，不会再重复读取源路径下的parquet文件。

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