PySpark DataFrame.where过滤嵌套数组列时丢失行的问题

问题重现 #

你需要移除HVA_label_list（嵌套数组类型）长度为0的行，通过Pandas统计得到预期32913行，但使用size(col("HVA_label_list")) > 0过滤后仅得到10636行，explode操作也出现类似数据缺失。

解决方案 #

改用更可靠的数组非空判断方式，同时处理null值：

方法1：使用isEmpty()结合非空校验 #

from pyspark.sql.functions import col

# 过滤掉null或空数组的行
result2 = result.filter(col("HVA_label_list").isNotNull() & ~col("HVA_label_list").isEmpty())
result2_pd = result2.toPandas()
print(len(result2_pd))  # 应返回预期的32913行

方法2：直接对比空数组 #

from pyspark.sql.functions import array, col

result2 = result.filter(col("HVA_label_list").isNotNull() & (col("HVA_label_list") != array()))

原因分析 #

1. null值处理差异：Spark中size()函数对null值的列返回null，null > 0的结果为null，where子句会自动过滤这些行；但你将Spark DataFrame转Pandas时，null可能被转换为空数组，导致Pandas统计的非空行数和Spark过滤结果出现偏差。
2. 嵌套数组的size()函数潜在问题：部分Spark版本中，size()对嵌套数组的判断可能出现异常（比如误判内层数组长度），改用直接判断数组是否为空的方式更稳定。

验证差异行 #

如果需要定位具体缺失行的原因，可以抽样检查：

# 提取Pandas中非空的customer_id集合
non_empty_pd_ids = set(result_pd[result_pd['HVA_label_list'].apply(len) > 0]['customer_id'])
# 提取Spark过滤后的customer_id集合
filtered_spark_ids = set(result2.select("customer_id").rdd.flatMap(lambda x: x).collect())
# 找出缺失的ID
missing_ids = list(non_empty_pd_ids - filtered_spark_ids)

# 查看这些ID在Spark中的数组状态
result.filter(col("customer_id").isin(missing_ids))\
     .select(
         "customer_id", 
         col("HVA_label_list"), 
         size(col("HVA_label_list")).alias("spark_size"), 
         col("HVA_label_list").isEmpty().alias("is_empty"),
         col("HVA_label_list").isNull().alias("is_null")
     )\
     .show(truncate=False)

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