如何在PySpark中访问_corrupt_record列处理CSV坏记录

问题原因 #

该错误由Spark SQL的谓词下推优化机制导致：当你直接对刚读取生成的DataFrame执行filter操作时，Spark优化器会尝试将过滤条件下推到CSV数据源的读取阶段，而_corrupt_record是CSV解析过程中生成的衍生列，不属于CSV文件的原生字段，读取阶段无法识别该列，因此抛出列不存在的错误。

可行解决方法 #

方法1：缓存DataFrame打断谓词下推 #

读取数据后先将DataFrame缓存到内存，切断谓词下推到数据源的链路，再执行过滤操作即可：

schema = StructType([
  StructField("Name",StringType(),True),
  StructField("ID",IntegerType(),True),
  StructField("int",IntegerType(),True),
  StructField("int2",IntegerType(),True),
  StructField("_corrupt_record", StringType(),True) 
])
df = spark.read.csv(
    'dbfs:/tmp/test_file/test_csv.csv', 
    header=True, 
    schema=schema, 
    columnNameOfCorruptRecord='_corrupt_record',
    mode='PERMISSIVE' # 显式指定宽容模式，默认也是该配置，显式声明更稳妥
)
# 先缓存数据，再执行过滤
df.cache()
badRows = df.filter("_corrupt_record is Not Null")
badRows.show()

方法2：用查询Hint禁用谓词下推 #

如果不想额外占用内存缓存数据，可以通过查询Hint直接禁用本次查询的过滤下推逻辑：

badRows = df.hint("DISABLE_PUSHDOWN_FILTER").filter("_corrupt_record is Not Null")
badRows.show()

方法3：全局关闭CSV读取的谓词下推（不推荐，会影响整体CSV读取性能） #

如果需要频繁处理带坏记录的CSV文件，可以在当前Spark会话临时调整配置：

spark.conf.set("spark.sql.sources.pushdown.csvFilters.enabled", "false")
# 调整配置后可直接执行过滤
badRows = df.filter("_corrupt_record is Not Null")
badRows.show()

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