Python使用pandas.to_csv写CSV时程序意外终止致文件清空如何解决

直接调用pandas.to_csv()写入目标文件时，程序打开文件的瞬间就会按写入模式清空原有内容（覆写模式），如果写入过程中程序意外终止，就会出现空文件或者半截内容的问题；并行场景下如果多个进程同时操作同一个文件且未加锁，会进一步放大该问题。

核心思路是用临时文件原子替换逻辑，操作系统会保证替换操作的原子性，只要替换没执行，原文件不会有任何改动，完全满足“要么写完要么保留原有状态”的要求。

通用安全写入函数实现 #

import os
import tempfile
import pandas as pd

def safe_append_df_to_csv(df, target_path, **to_csv_kwargs):
    # 临时文件与目标文件放在同一目录，保证原子替换生效
    target_dir = os.path.dirname(os.path.abspath(target_path))
    # 创建临时文件，手动控制删除逻辑
    with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', dir=target_dir, delete=False) as tmp_f:
        tmp_path = tmp_f.name

    try:
        if os.path.exists(target_path):
            # 原文件存在时先复制历史内容到临时文件
            with open(target_path, 'r', encoding='utf-8') as src_f:
                with open(tmp_path, 'w', encoding='utf-8') as tmp_write_f:
                    tmp_write_f.write(src_f.read())
            # 追加新数据，不重复写表头
            df.to_csv(tmp_path, mode='a', header=False, index=False, **to_csv_kwargs)
        else:
            # 原文件不存在时直接写新数据，带表头
            df.to_csv(tmp_path, mode='w', header=True, index=False, **to_csv_kwargs)
        
        # 强制刷盘，确保所有数据写入磁盘
        tmp_fd = os.open(tmp_path, os.O_RDONLY)
        os.fsync(tmp_fd)
        os.close(tmp_fd)
        # 原子替换原文件，兼容Windows和Unix系统
        os.replace(tmp_path, target_path)
    finally:
        # 无论是否成功，都清理残留的临时文件
        if os.path.exists(tmp_path):
            os.remove(tmp_path)

并行场景适配 #

• 若每个进程操作独立的CSV文件，直接调用上述函数即可，无需额外处理
• 若多个进程需要向同一个CSV追加数据，需要加文件锁避免内容冲突，可以用第三方库filelock简化实现：

from filelock import FileLock

def parallel_safe_append(df, target_path, **to_csv_kwargs):
    # 同目录下创建锁文件，避免多进程同时写入
    lock = FileLock(f"{target_path}.lock")
    with lock:
        safe_append_df_to_csv(df, target_path, **to_csv_kwargs)

大文件场景性能优化方案 #

如果CSV文件过大，每次追加都复制全量历史内容性能太低，可以选择刷盘优化的追加方案，该方案不会清空历史内容，极端情况只会丢失本次追加的部分数据，性能远高于原子替换方案：

def fast_append_csv(df, target_path, **to_csv_kwargs):
    write_header = not os.path.exists(target_path)
    with open(target_path, 'a', encoding='utf-8') as f:
        df.to_csv(f, header=write_header, index=False, **to_csv_kwargs)
        # 强制刷盘，避免数据留在操作系统缓存中
        f.flush()
        os.fsync(f.fileno())

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