Pandas追加写CSV如何处理异字段API数据避免列错位与重复表头

核心逻辑是维护全局字段列表，每次新批次返回数据后对齐字段，仅在出现新增字段时重写全量文件，其余场景增量追加，从根源避免列错位问题。

完整实现代码 #

import pandas as pd
import os

csv_path = "Data.csv"
all_columns = []
# 初始化全局字段列表，仅读取表头不加载全量数据，性能开销极低
if os.path.exists(csv_path):
    all_columns = pd.read_csv(csv_path, nrows=0).columns.tolist()

allFetched = False
while not allFetched:
    response = callApi()
    # 生成当前批次DataFrame
    new_df = pd.DataFrame.from_dict(response['data'])
    # 计算新增字段，保留原有字段顺序，新增字段追加到末尾
    new_columns = [col for col in new_df.columns if col not in all_columns]
    current_all_columns = all_columns + new_columns

    if len(new_columns) > 0 and os.path.exists(csv_path):
        # 存在新增字段，重写全量文件保证历史数据也有新字段列
        old_df = pd.read_csv(csv_path)
        # 新旧数据统一对齐全量字段，缺失值自动填充为NaN
        old_df = old_df.reindex(columns=current_all_columns)
        new_df = new_df.reindex(columns=current_all_columns)
        # 合并后覆盖写入原文件
        full_df = pd.concat([old_df, new_df], ignore_index=True)
        full_df.to_csv(csv_path, index=False, header=True)
        # 更新全局字段列表
        all_columns = current_all_columns
    else:
        # 无新增字段，直接追加写入即可
        write_columns = all_columns if all_columns else current_all_columns
        new_df = new_df.reindex(columns=write_columns)
        # 仅首次写入时输出表头
        need_header = not os.path.exists(csv_path)
        new_df.to_csv(csv_path, mode='a', index=False, header=need_header)
        # 首次写入后初始化全局字段列表
        if not all_columns:
            all_columns = current_all_columns
    
    # 此处替换为你自己的分页终止判断逻辑
    # allFetched = 判断是否还有下一页数据

关键逻辑说明 #

• 首次写入文件时自动生成当前全量字段的表头，后续追加不会重复写入表头
• 无新增字段时仅做增量追加，不需要读取和重写历史数据，性能几乎和原始实现一致
• 出现新增字段时自动补全所有历史行的新增字段值为NaN，保证所有行列数、列顺序完全一致，不会出现错位
• 新增字段默认追加到已有字段末尾，不会打乱原有列的顺序

大数据量场景优化（可选） #

如果拉取的数据量级极大，重写全量文件成本过高，可采用以下方案降低性能开销：

• 先将每次拉取的原始数据以JSON Lines格式逐行写入.jsonl文件，无需提前对齐字段
• 所有数据拉取完成后，一次性读取全量JSON Lines文件转成DataFrame，自动补全所有缺失字段后统一导出为CSV，仅需一次全量计算即可完成字段对齐

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