如何使用pandas.read_csv()处理畸形CSV并拆分得到两个DataFrame

pd.read_csv() 原生不支持单次调用直接返回两个DataFrame，这个函数的设计返回值固定为单个DataFrame（或者分块读取的TextFileReader对象），无法在函数内部直接拆分输出两张表。但可以通过on_bad_lines参数配合闭包列表捕获的方式，实现纯Pandas实现的单次读入、自动拆分正常/异常数据，不需要手写逐行遍历清洗的逻辑，完全适配你的数据规模。

• 初始化一个独立列表，专门存储解析过程中识别到的异常行元数据
• 编写自定义的坏行处理函数传入on_bad_lines：函数拿到字段数不匹配的畸形行拆分结果后，先通过你提到的failed关键词定位错误字段，提取错误信息、错误位置、关联ID等信息存入异常列表；再给残缺行补全空值到和正常表一致的字段长度，保证Pandas可以正常完成解析流程不中断
• 读入完成后，将存储异常记录的列表直接转为错误编目用的DataFrame；再从主表中过滤掉存在错误的记录，就得到供仪表盘使用的干净指标表

import pandas as pd

# 定义正常记录的14个字段名，可根据实际业务替换
NORMAL_COLUMNS = [
    "unique_id", "create_date", "exec_time", "customer_id", "task_type",
    "metric_5", "metric_6", "metric_7", "metric_8", "metric_9",
    "metric_10", "metric_11", "metric_12", "metric_13"
]
# 存储异常记录的列表
error_collection = []

def parse_bad_line(line_fields: list[str]) -> list[str]:
    # 定位错误信息所在字段
    error_msg = None
    error_col_index = None
    for idx, field in enumerate(line_fields):
        if "failed" in field:
            error_msg = field.strip('"')
            error_col_index = idx
            break
    # 存入异常记录集合
    error_collection.append({
        "unique_id": line_fields[0],
        "error_column_pos": error_col_index,
        "error_detail": error_msg,
        "raw_content": ",".join(line_fields)
    })
    # 补全字段长度到14位，缺省位置填空值，保证read_csv正常解析
    fixed_line = line_fields[:error_col_index] + [None] * (len(NORMAL_COLUMNS) - error_col_index)
    return fixed_line

# 读取数据，注意必须用python引擎才支持自定义on_bad_lines函数
df_all = pd.read_csv(
    "your_process_log.csv",
    header=None,
    names=NORMAL_COLUMNS,
    on_bad_lines=parse_bad_line,
    engine="python"
)

# 先过滤字段数不足触发回调的异常行，再过滤字段数完整但最后一位是错误信息的行
df_clean = df_all[~df_all["unique_id"].isin([item["unique_id"] for item in error_collection])]
full_line_error_mask = df_clean["metric_13"].str.contains("failed", na=False)
full_line_errors = df_clean[full_line_error_mask]
df_clean = df_clean[~full_line_error_mask].reset_index(drop=True)

# 补充收集字段数完整的错误行（比如示例里id2这类14个字段、但最后一位是错误信息的记录）
if not full_line_errors.empty:
    for _, row in full_line_errors.iterrows():
        error_collection.append({
            "unique_id": row["unique_id"],
            "error_column_pos": 13,
            "error_detail": row["metric_13"],
            "raw_content": ",".join([str(row[col]) for col in NORMAL_COLUMNS])
        })
# 生成最终的错误信息表，可直接用于错误分类编目
df_errors = pd.DataFrame(error_collection)

• 必须指定engine="python"，Pandas的C解析引擎不支持传入自定义可调用对象作为on_bad_lines参数
• 该方案完全走Pandas内置的CSV解析流程，性能远高于手写逐行读取清洗，针对你月均数千条的数据规模几乎没有性能压力
• 可以在parse_bad_line函数里直接加错误分类逻辑，比如识别到DELETE error就打标为"删除操作失败"、识别到NO SUCH JOB就打标为"任务不存在"，不需要读入后再二次处理

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