PySpark中捕获Spark CSV读取异常的可行方案咨询

问题 #

在Databricks笔记本中，读取CSV文件是ETL流水线的第一步。当CSV文件不符合指定Schema时，会导致笔记本崩溃，需要在错误发生时处理而非让整个流水线终止。

尝试用以下代码捕获异常，期望读取错误CSV时输出Exception caught：

try:
    newData = (spark.read
              .format("csv")
              .option("delimiter", "|")
              .option("mode", "FAILFAST")
              .option("inferSchema", "false")
              .option("enforceSchema", "true")
              .option("header", "True")
              .schema(schema)
              .load(bronzePath + "/"+ fileName + "*")
         )
    newData.display()
except FileReadException as e:
    # Do stuff, handle exception
    print("Exception caught")

但异常未被捕获，反而输出完整异常栈：

FileReadException: Error while reading file dbfs:/mnt/datalake/bronze/<myPath>/<myFile.csv>.
Caused by: SparkException: Malformed records are detected in record parsing. Parse Mode: FAILFAST. To process malformed records as null result, try setting the option 'mode' as 'PERMISSIVE'.
Caused by: BadRecordException: org.apache.spark.sql.catalyst.csv.MalformedCSVException: Malformed CSV record
Caused by: MalformedCSVException: Malformed CSV record

已知PySpark无法直接捕获Scala抛出的FileReadException，需要其他方法捕获该异常，且不能使用PERMISSIVE或DROPMALFORMED模式，必须对错误CSV文件进行处理。

解决方案 #

1. 捕获Py4J包装的Scala异常 #

PySpark调用Scala底层代码时，Scala异常会被包装为Py4JJavaError。可以捕获该异常，并通过检查异常信息中的关键词判断是否为目标错误：

from py4j.protocol import Py4JJavaError

try:
    newData = (spark.read
              .format("csv")
              .option("delimiter", "|")
              .option("mode", "FAILFAST")
              .option("inferSchema", "false")
              .option("enforceSchema", "true")
              .option("header", "True")
              .schema(schema)
              .load(bronzePath + "/"+ fileName + "*")
         )
    newData.display()
except Py4JJavaError as e:
    java_exception = str(e.java_exception)
    if "MalformedCSVException" in java_exception or "FileReadException" in java_exception:
        print("Exception caught")
        # 添加错误文件处理逻辑，例如移动到错误目录
        dbutils.fs.mv(bronzePath + "/"+ fileName + "*", "/mnt/datalake/error_files/" + fileName)
except Exception as e:
    print(f"Unexpected error: {str(e)}")

2. 预校验文件结构 #

在正式读取全量数据前，先读取少量样本数据进行初步校验，提前发现Schema不匹配问题：

# 读取前10行样本数据
sample_df = (spark.read
             .format("csv")
             .option("delimiter", "|")
             .option("header", "True")
             .option("inferSchema", "false")
             .load(bronzePath + "/"+ fileName + "*")
             .limit(10)
            )

# 校验列数是否匹配目标Schema
if len(sample_df.columns) != len(schema.fields):
    print("Exception caught: Column count mismatch")
    dbutils.fs.mv(bronzePath + "/"+ fileName + "*", "/mnt/datalake/error_files/" + fileName)
else:
    # 列数匹配后，再用FAILFAST模式读取全量数据
    try:
        newData = (spark.read
                  .format("csv")
                  .option("delimiter", "|")
                  .option("mode", "FAILFAST")
                  .option("inferSchema", "false")
                  .option("enforceSchema", "true")
                  .option("header", "True")
                  .schema(schema)
                  .load(bronzePath + "/"+ fileName + "*")
             )
        newData.display()
    except Py4JJavaError as e:
        if "MalformedCSVException" in str(e.java_exception):
            print("Exception caught: Malformed CSV record")
            dbutils.fs.mv(bronzePath + "/"+ fileName + "*", "/mnt/datalake/error_files/" + fileName)

3. 逐行检查文件内容 #

利用Databricks的文件工具dbutils读取文件内容，逐行校验格式（如分割符数量），提前拦截错误文件：

file_list = dbutils.fs.ls(bronzePath + "/"+ fileName + "*")

for file in file_list:
    # 读取文件前10KB内容（可根据需求调整）
    file_content = dbutils.fs.head(file.path, 10000)
    lines = file_content.split("\n")
    
    # 跳过空行和表头，检查数据行格式
    valid = True
    for line in lines[1:]:
        line = line.strip()
        if not line:
            continue
        # 校验分割符数量是否等于Schema列数-1
        if line.count("|") != len(schema.fields) - 1:
            valid = False
            break
    
    if not valid:
        print("Exception caught: Invalid line format")
        dbutils.fs.mv(file.path, "/mnt/datalake/error_files/" + file.name)
    else:
        # 格式校验通过后读取全量数据
        try:
            newData = (spark.read
                      .format("csv")
                      .option("delimiter", "|")
                      .option("mode", "FAILFAST")
                      .option("inferSchema", "false")
                      .option("enforceSchema", "true")
                      .option("header", "True")
                      .schema(schema)
                      .load(file.path)
                 )
            newData.display()
        except Py4JJavaError as e:
            if "MalformedCSVException" in str(e.java_exception):
                print("Exception caught: Malformed CSV record")
                dbutils.fs.mv(file.path, "/mnt/datalake/error_files/" + file.name)

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