PySpark Filter仅缓存后生效问题：MongoDB加载DF过滤报错求解

核心原因 #

Spark采用懒执行机制：

• show()和select()默认仅处理小批量数据（比如show()默认展示20行），你测试的这批数据恰好都有完整的message.data.payload.ChangeEventHeader.changeType嵌套结构，因此能正常返回结果。
• 而filter()需要遍历全量数据，当遇到MongoDB中缺失该嵌套字段的文档时，就会抛出字段不存在的错误。

缓存方案的合理性 #

缓存是合理的，但并非最优解：

1. 缓存会触发Spark立即执行全量数据计算，提前将缺失字段的位置填充为null，后续filter操作直接读取缓存数据，不会重新扫描原MongoDB数据，因此不会报错。
2. 但缓存会占用集群内存，若数据量极大，可能引发内存压力。

更稳妥的替代方案 #

直接在提取字段时处理缺失情况，无需依赖缓存：

from pyspark.sql.functions import col, coalesce, lit, when

# 方案1：用coalesce处理缺失，缺失时填充null
df = df.withColumn(
    'ctype',
    coalesce(
        col('message.data.payload.ChangeEventHeader.changeType'),
        lit(None)
    )
)

# 方案2：逐层判断字段是否存在，避免深层嵌套报错
df = df.withColumn(
    'ctype',
    when(
        col('message').isNotNull() & 
        col('message.data').isNotNull() & 
        col('message.data.payload').isNotNull() & 
        col('message.data.payload.ChangeEventHeader').isNotNull(),
        col('message.data.payload.ChangeEventHeader.changeType')
    ).otherwise(lit(None))
)

df.filter(col('ctype') == "AAA").show()

这种方式能提前处理所有可能的字段缺失情况，避免全量缓存带来的内存问题，同时保证后续过滤操作稳定执行。

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