Spark过滤损坏记录字段时.count()函数结果与DataFrame内容不符

我之前也碰到过类似的头疼问题，这种不一致通常和Spark的核心运行机制或者数据读取配置有关，咱们一步步来排查解决：

1. 先确认坏记录的标记逻辑是否正确 #

首先得确保Spark在读取数据时，确实把不符合Schema的损坏记录都放到了你指定的字段里。默认情况下，Spark的PERMISSIVE模式（CSV/JSON等数据源的默认模式）会把不匹配的记录写入_corrupt_record字段，但如果没显式配置相关参数，或者Schema定义有疏漏，可能导致坏记录没被正确标记。

比如读取CSV时，建议明确配置这些参数：

df = spark.read.schema(SCHEMA) \
    .option("mode", "PERMISSIVE") \
    .option("columnNameOfCorruptRecord", "_corrupt_record") \
    .csv("your_input_file_path")

注意columnNameOfCorruptRecord要和你过滤时用的字段名完全一致，避免因字段名不匹配导致过滤失效。

2. 警惕Spark的延迟计算特性 #

Spark是延迟计算的，所有filter这类转换操作不会立即执行，只有调用count()、show()这类行动操作时才会触发整个作业的计算。如果你的过滤逻辑之后还叠加了其他转换，可能因为依赖链的问题导致计算结果和预期不符。

可以先把过滤后的DataFrame缓存起来，再执行统计：

valid_df = df.filter(col("_corrupt_record").isNull())
valid_df.cache()  # 缓存过滤后的结果，避免重复计算
print(valid_df.count())
valid_df.show(50, truncate=False)  # 直接对比显示内容和统计数

如果缓存后还是不一致，那大概率是数据读取阶段的配置出了问题。

3. 排查数据分区与执行日志 #

如果count()和实际显示的内容对不上，建议打开Spark UI（默认地址是http://localhost:4040）查看作业详情：

• 看每个任务的输入记录数，是否有分区的记录数异常
• 检查是否有任务失败重试的情况，部分重试可能导致重复统计
• 查看日志里是否有关于坏记录处理的警告信息，比如Schema不匹配的提示

另外，如果你用的是local模式，可以暂时改成local[1]，避免多线程执行时的潜在冲突。

4. 验证过滤逻辑的准确性 #

有时候可能是过滤条件写反了，比如想保留有效记录却写成了保留坏记录：

# 错误：保留了损坏记录
invalid_df = df.filter(col("_corrupt_record").isNotNull())
# 正确：保留有效记录
valid_df = df.filter(col("_corrupt_record").isNull())

可以先通过distinct()查看过滤后坏记录字段的取值：

valid_df.select("_corrupt_record").distinct().show()

如果结果里还有非空值，说明过滤逻辑没生效，得重新检查条件。

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