PySpark读取CSV错误记录处理：避免重复读取的优化方案问询

针对你遇到的重复读取文件、DAG重复执行问题，结合Unity Catalog环境，以下是几个比persist(MEMORY_AND_DISK)更优的方案：

方案1：一次读取分离正确/错误记录，避免重复扫描 #

利用Spark的PERMISSIVE模式读取CSV，通过自定义校验逻辑识别错误记录（替代columnNameOfCorruptedRecord），一次读取就拆分出有效数据和错误数据，彻底避免重复读取文件。

实现步骤： #

• 读取CSV时开启PERMISSIVE模式，保留原始行数据
• 定义校验规则（比如字段数量匹配、关键字段非空、数据类型合规）筛选错误记录
• 检查错误记录数量，非零则抛出异常；否则将有效数据追加到bronze_table

from pyspark.sql.functions import split, size, input_file_name, current_timestamp

# 定义CSV预期字段数
expected_cols_count = 5

# 一次读取CSV，保留原始行及元数据
raw_df = (spark.read
          .option("mode", "PERMISSIVE")
          .option("header", "true")
          .option("delimiter", ",")
          .csv("/path/to/your/csv")
          .withColumn("_raw_file", input_file_name())
          .withColumn("_load_timestamp", current_timestamp()))

# 自定义逻辑筛选错误记录：字段数不符或关键字段为空
error_records_df = raw_df.filter(
    size(split(raw_df["_raw_file"], ",")) != expected_cols_count 
    | raw_df["critical_field"].isNull()
)

# 检查错误记录
error_count = error_records_df.count()
if error_count > 0:
    # 将错误记录写入Unity Catalog下的错误日志表（可选）
    error_records_df.write.mode("append").saveAsTable("uc_catalog.uc_schema.error_log_table")
    raise Exception(f"Found {error_count} corrupted records, check error_log_table for details")

# 筛选有效数据并追加到bronze_table
valid_df = raw_df.exceptAll(error_records_df)
valid_df.write.mode("append").saveAsTable("uc_catalog.uc_schema.bronze_table")

优势：仅扫描一次CSV文件，DAG执行路径简洁；自定义校验逻辑灵活，能精准捕获错误场景；错误记录可持久化到Unity Catalog的日志表，方便后续排查。

方案2：使用Delta Lake COPY INTO（Databricks原生最优方案） #

在Unity Catalog环境下，COPY INTO是专为批量数据 ingestion 设计的原生API，自带错误处理机制，无需手动管理重复读取，且支持原子性操作，完全适配Unity Catalog权限模型。

实现步骤： #

• 创建Delta格式的错误日志表（存储错误记录和详情）
• 执行COPY INTO时指定错误处理策略，将错误路由到日志表
• 检查错误日志表的新增记录数，非零则抛出异常

-- 1. 在Unity Catalog下创建错误日志表（仅需执行一次）
CREATE TABLE IF NOT EXISTS uc_catalog.uc_schema.csv_error_log (
  raw_data STRING,
  error_message STRING,
  load_timestamp TIMESTAMP
) USING DELTA;

-- 2. 执行COPY INTO，将错误记录写入日志表
COPY INTO uc_catalog.uc_schema.bronze_table
FROM '/path/to/your/csv'
FILEFORMAT = CSV
OPTIONS (header = 'true', delimiter = ',')
COPY_OPTIONS (
  mergeSchema = 'true',
  onError = 'continue' -- 继续处理，自动将错误记录路由到日志表
)
ERROR LOG INTO uc_catalog.uc_schema.csv_error_log;

-- 3. 检查本次加载的错误记录数
DECLARE error_count INT;
SET error_count = (
  SELECT COUNT(*) 
  FROM uc_catalog.uc_schema.csv_error_log
  WHERE load_timestamp >= CURRENT_TIMESTAMP() - INTERVAL 5 MINUTE -- 过滤本次加载的错误
);

IF error_count > 0 THEN
  RAISE EXCEPTION 'Found % corrupted records, check csv_error_log for details', error_count;
END IF;

优势：Databricks原生优化，性能远超手动读取；自动处理重复文件（避免重复加载）；错误日志结构化存储，便于后续分析；无需手动管理数据拆分，代码更简洁。

方案3：优化持久化策略（替代MEMORY_AND_DISK） #

如果坚持使用持久化，在大数据量场景下，优先选择DISK_ONLY存储级别，彻底避免内存占用问题，同时结合资源释放确保稳定性：

from pyspark import StorageLevel

raw_df = (spark.read
          .option("mode", "PERMISSIVE")
          .option("header", "true")
          .csv("/path/to/your/csv"))

# 使用DISK_ONLY持久化，仅写入磁盘，无内存压力
raw_df.persist(StorageLevel.DISK_ONLY)

# 自定义逻辑筛选错误记录
error_records_df = raw_df.filter(/* 错误校验规则 */)
error_count = error_records_df.count()

if error_count > 0:
    error_records_df.write.mode("append").saveAsTable("uc_catalog.uc_schema.error_log_table")
    raise Exception(f"Found {error_count} corrupted records")

# 写入有效数据到bronze_table
raw_df.filter(/* 有效数据规则 */).write.mode("append").saveAsTable("uc_catalog.uc_schema.bronze_table")

# 释放持久化资源
raw_df.unpersist()

优势：比MEMORY_AND_DISK更稳定，适合超大数据集；持久化后仅扫描一次文件，解决DAG重复执行问题。

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