Pyspark Struct列执行explode操作后出现取值异常问题

异常原因 #

该列值变化不属于Spark的正常运行逻辑。explode仅会对指定的数组列执行行展开，其余已经生成的列会原样复制到展开后的每一行，不会主动修改取值。出现异常通常为三类原因：

• 原始数据中values_level1.values存在空数组，默认的explode会直接过滤空数组对应的整行，导致你误以为对应name_level1的取值发生了变化
• 中间结果未缓存，Spark的懒加载机制导致多次重算时出现列引用错位，或读取到了发生变更的原始数据源
• 嵌套字段的引用路径填写错误，取到了其他层级的name字段值

修复方案 #

方案1：保留空数组对应行 #

用explode_outer替代explode，空数组对应的行不会被过滤，仅values_level2字段会填充为null，name_level1的值会正常保留：

jsonDF = jsonDF.withColumn("values_level2", explode_outer("values_level1.values"))

方案2：缓存中间结果避免重算异常 #

第一次展开并提取name_level1后，先缓存中间结果再执行第二层展开，确保所有操作基于稳定的中间数据执行：

jsonDF = jsonDF.withColumn("values_level1", explode("data.values"))
               .withColumn("name_level1", col("values_level1.name"))
               .cache()

// 执行第二层展开
jsonDF = jsonDF.withColumn("values_level2", explode("values_level1.values"))

方案3：调整字段提取顺序 #

第二层展开完成后再显式提取name_level1，彻底避免列引用错位问题：

jsonDF = jsonDF.withColumn("values_level1", explode("data.values"))
               .withColumn("values_level2", explode("values_level1.values"))
               .withColumn("name_level1", col("values_level1.name"))

问题示意图 #

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