PySpark读取CSV设置DROPMALFORMED模式后无返回结果问题

问题现象 #

指定存储路径下的CSV文件本身包含有效数据，但执行读取代码后未返回任何结果：

• 初始使用ValidFile = spark.read.csv(ValidationFileDest, header = True)读取时可正常返回结果，但存在列数据错位、空值赋值异常的问题
• 在读取配置中添加DROPMALFORMED模式后，执行查询无任何结果返回

复现代码 #

parquetextension=".parquet"
BronzeStage_Path = "dbfs:/mnt/bronze/stage/" +parentname+"/" +filename
#validated_path="dbfs:/mnt/bronze/landing/ClaimDenialsSouce/"+parentname+"/"+"current/"+"Valid/"+todayDate+"_"+"CDAValidFile"+extension
# df_sourcefilevalid.repartition(1).write.format(write_format).option("header", "true").save(BronzeStagePath)
# ValidFileSrc_BS= get_csv_files(exception_path)
from pyspark.sql import SparkSession
spark = SparkSession.builder \
  .master("local") \
  .appName("parquet_example") \
  .getOrCreate()
spark.conf.set("spark.sql.csv.parser.columnPruning.enabled",False)
ValidFile = spark.read.format('csv').option("mode","DROPMALFORMED").options(header='true', inferSchema='true').load(ValidationFileDest)
display(ValidFile)

问题根因 #

DROPMALFORMED模式的判定规则为：某行解析出的字段数量与schema列数不匹配时，直接丢弃整行。
之前出现列错位、空值异常，本质是CSV文件存在以下格式问题：字段内含未转义的逗号、字段内存在未被正确识别的换行符、部分行存在多余/缺失分隔符，这些问题会导致Spark解析时判定所有行的字段数和表头推导出的schema不匹配，开启DROPMALFORMED后所有行被全量丢弃，因此返回空结果。

修复方案 #

不要直接使用DROPMALFORMED全量丢弃疑似异常行，按以下步骤调整：

1. 补全CSV解析参数，适配特殊字符和多行字段场景

ValidFile = spark.read.format("csv")\
    .option("header", "true")\
    .option("inferSchema", "true")\
    .option("quote", "\"")\
    .option("escape", "\"")\
    .option("multiLine", "true")\
    .option("mode", "PERMISSIVE")\
    .load(ValidationFileDest)

如果CSV实际分隔符不是逗号，额外添加.option("sep", "实际分隔符")配置，比如分号分隔就填";"
2. 读取后分离正常数据和异常数据，单独排查异常行而非全量丢弃

from pyspark.sql.functions import col, isnull
# PERMISSIVE模式下会把解析异常的行放到_corrupt_record字段
normal_df = ValidFile.filter(isnull(col("_corrupt_record"))).drop("_corrupt_record")
corrupt_df = ValidFile.filter(~isnull(col("_corrupt_record")))

3. 校验正常数据的列错位、空值问题是否解决，异常行可单独导出后排查原始格式问题。

相关佐证材料 #

• 查询结果输出截图
• CSV文件原始数据截图

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