Pandas向CSV追加数据：追加前检查文件是否存在表头

这个问题在分布式/多进程写CSV场景下太常见了，我之前做批量数据处理的时候也踩过同样的坑！给你几个不同场景下的实用方案：

1. 单机器多进程场景：用文件锁保证表头只写一次 #

当所有Worker是同一台机器上的进程时，我们可以借助文件锁实现原子性的检查和写入，避免多个进程同时判断文件不存在而重复写入表头。

代码示例：

import os
import fcntl
import pandas as pd

def append_df_to_csv(df, csv_path):
    file_exists = os.path.isfile(csv_path)
    with open(csv_path, 'a') as f:
        # 获取排他锁，同一时间只有一个进程能操作文件
        fcntl.flock(f, fcntl.LOCK_EX)
        try:
            if not file_exists:
                # 第一次写入，带完整表头
                df.to_csv(f, index=False)
            else:
                # 后续追加，只写数据行
                df.to_csv(f, index=False, header=False)
        finally:
            # 无论成功失败都释放锁，避免死锁
            fcntl.flock(f, fcntl.LOCK_UN)

注意：Windows系统下需要替换fcntl为msvcrt模块的锁函数，逻辑完全一致。

2. 跨机器分布式场景：原子操作创建表头标记 #

如果Worker在不同机器，共享存储（比如NFS、S3），普通文件锁可能不可靠，这时候可以用原子性的文件创建操作来标记表头是否已写入：

本地共享存储（如NFS） #

import os
import pandas as pd

def append_df_to_csv_distributed(df, csv_path, header_flag="header_written.flag"):
    try:
        # 原子创建标记文件，只有第一个Worker能成功
        os.mknod(header_flag)
        # 标记创建成功，写入表头+数据
        df.to_csv(csv_path, index=False)
    except FileExistsError:
        # 标记已存在，只追加数据行
        df.to_csv(csv_path, mode='a', index=False, header=False)
    
    # 所有任务完成后可以删除标记文件
    # if all_tasks_completed: os.remove(header_flag)

os.mknod是原子操作，多个Worker同时调用时只有一个会成功，完美避免重复写表头。

对象存储（如S3） #

S3对象是不可变的，无法直接追加，这种场景下更推荐下面的「事后合并」方案，能避免复杂的多版本合并逻辑。

3. 最稳妥的方案：先写独立文件，最后统一合并 #

如果你的任务允许事后处理，强烈推荐这种方式：每个Worker生成自己的独立CSV文件，等所有任务完成后，再统一合并成带表头的最终文件。

Worker端代码 #

# 每个Worker生成唯一命名的CSV，比如用Worker ID或任务ID区分
df.to_csv(f'data_output_{worker_id}.csv', index=False)

合并端代码 #

import pandas as pd
import glob
import os

# 收集所有Worker生成的CSV文件
all_data_files = glob.glob('data_output_*.csv')

# 合并所有DataFrame
combined_df = pd.concat(
    [pd.read_csv(file) for file in all_data_files],
    ignore_index=True
)

# 写入最终CSV（只写一次表头）
combined_df.to_csv('test.csv', index=False)

# 可选：清理临时文件
for file in all_data_files:
    os.remove(file)

这种方式完全避免了并发写入的冲突，还能方便地做数据校验，适合大规模的分布式任务。

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