Polars read_csv遇分隔符错误：如何统计/导出无效记录？

方法一：利用on_bad_lines自定义错误处理 #

Polars的read_csv提供了on_bad_lines参数（替代旧版ignore_errors），可以自定义函数处理错误行，同时实现统计无效记录数和导出无效行到文件的需求。

示例代码：

import polars as pl

# 初始化统计变量和文件句柄
bad_line_count = 0
with open("invalid_records.csv", "w", encoding="utf-8") as bad_file:
    def handle_bad_lines(line: str) -> None:
        global bad_line_count
        bad_line_count += 1
        # 将错误行写入指定文件
        bad_file.write(line + "\n")
        # 返回None表示跳过该行
        return None

    # 读取CSV并绑定错误处理函数
    df = pl.read_csv(
        "large_dataset.csv",
        separator=",",  # 替换为你的实际分隔符
        on_bad_lines=handle_bad_lines
    )

print(f"共跳过 {bad_line_count} 条无效记录")

该方法优势：

• 实时捕获每一行错误，统计精准
• 直接保留错误行原始格式并导出
• 无需额外全量扫描，性能适配大型数据集

方法二：逐行预校验（适配复杂分隔符场景） #

如果字段内包含未正确转义的分隔符，导致规则异常，可以先通过逐行校验筛选有效行，再交给Polars读取：

import polars as pl

expected_columns = 5  # 替换为你的预期字段数量
bad_lines = []
good_lines = []

# 逐行读取并校验字段数
with open("large_dataset.csv", "r", encoding="utf-8") as f:
    # 先读取表头
    header = next(f)
    good_lines.append(header)
    for line in f:
        split_line = line.strip().split(",")  # 替换为你的分隔符
        if len(split_line) == expected_columns:
            good_lines.append(line)
        else:
            bad_lines.append(line)

# 写入有效行到临时文件，再用Polars读取
with open("valid_temp.csv", "w", encoding="utf-8") as f:
    f.writelines(good_lines)
df = pl.read_csv("valid_temp.csv")

# 保存无效行
with open("invalid_records.csv", "w", encoding="utf-8") as f:
    f.writelines(bad_lines)

print(f"共跳过 {len(bad_lines)} 条无效记录")

注意：该方法需全量读取一次文件，超大型数据集（几十GB级）可改用分块读取优化内存占用。

方法三：流式扫描统计无效行数（仅需数量时使用） #

对于超大型数据集，用scan_csv流式处理，通过字段数匹配筛选有效行，同时统计无效行数：

import polars as pl

# 流式扫描获取预期字段数
schema = pl.scan_csv("large_dataset.csv").schema
expected_col_num = len(schema)

# 筛选有效行并收集
valid_df = (
    pl.scan_csv("large_dataset.csv", infer_schema_length=0)
    .filter(pl.len_horizontal(pl.all()) == expected_col_num)
    .collect()
)

# 计算无效行数：总行数 - 有效行数
total_rows = pl.scan_csv("large_dataset.csv").select(pl.count()).collect().item()
bad_line_count = total_rows - len(valid_df)

print(f"共跳过 {bad_line_count} 条无效记录")

该方法内存占用极低，但无法导出具体无效行内容，适合仅需统计数量的场景。

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