如何在Python中分割实时CSV日志文件并归档旧数据

核心思路 #

通过原子文件操作避免中断实时写入，将归档操作与实时写入解耦：先把正在写入的data.csv原子重命名为临时文件，立刻创建新的data.csv并写入表头，让原写入脚本无缝切换到新文件；之后在后台异步处理临时文件，分离出12小时内的有效数据写回新文件，旧数据归档到单独文件。整个过程中只有重命名+创建新文件的瞬间耗时可忽略，完全不影响每秒写入的业务。

步骤1：优化实时写入脚本 #

确保写入操作能自动适配文件变化，避免因文件重命名导致写入错位：

import time
import csv
import os

class DataWriter:
    def __init__(self, filename, header):
        self.filename = filename
        self.header = header
        self.file = None
        self.inode = None
        # 初始化时检查文件是否存在，不存在则写入表头
        if not os.path.exists(filename):
            with open(filename, 'w', newline='') as f:
                csv.writer(f).writerow(header)
        self._open_file()
    
    def _open_file(self):
        if self.file:
            self.file.close()
        # 以追加模式打开文件，行缓冲保证实时写入
        self.file = open(self.filename, 'a', newline='', buffering=1)
        self.writer = csv.writer(self.file)
        # 记录当前文件的inode（用于检测文件是否被替换）
        self.inode = os.stat(self.filename).st_ino
    
    def write_row(self, row):
        # 检查文件inode是否变化（原文件被重命名/替换）
        current_inode = os.stat(self.filename).st_ino
        if current_inode != self.inode:
            self._open_file()
        self.writer.writerow(row)
    
    def close(self):
        if self.file:
            self.file.close()

# 初始化写入器，替换成你的实际表头
header = ['epoch_time', 'int1', 'int2', 'int3']
writer = DataWriter('data.csv', header)

# 模拟每秒写入（替换成你的业务逻辑）
try:
    while True:
        current_epoch = int(time.time())
        data_row = [current_epoch, 1, 2, 3]
        writer.write_row(data_row)
        time.sleep(1)
finally:
    writer.close()

步骤2：归档脚本（定时执行） #

用定时任务（如每小时一次）执行归档，处理逻辑如下：

import os
import time
import csv
from datetime import datetime

def archive_old_data():
    # 计算12小时前的时间戳阈值
    threshold = time.time() - 12 * 3600
    temp_file = 'data_temp.csv'
    # 归档文件按时间命名，方便追溯
    archive_file = f'archive_{datetime.now().strftime("%Y%m%d%H")}.csv'
    
    # 原子重命名原文件为临时文件（操作系统级原子操作，无耗时）
    try:
        os.rename('data.csv', temp_file)
    except FileNotFoundError:
        # 若原文件刚创建为空，直接跳过
        return
    
    # 立刻创建新的data.csv并写入表头
    with open('data.csv', 'w', newline='') as f:
        csv.writer(f).writerow(['epoch_time', 'int1', 'int2', 'int3'])  # 替换成你的实际表头
    
    # 逐行处理临时文件，分离新/旧数据
    with open(temp_file, 'r', newline='') as temp_f, \
         open('data.csv', 'a', newline='') as new_data_f, \
         open(archive_file, 'a', newline='') as archive_f:
        
        reader = csv.reader(temp_f)
        new_writer = csv.writer(new_data_f)
        archive_writer = csv.writer(archive_f)
        
        # 跳过临时文件的表头
        next(reader)
        
        for row in reader:
            try:
                epoch = float(row[0])
                if epoch >= threshold:
                    # 保留12小时内的数据，写回新的data.csv
                    new_writer.writerow(row)
                else:
                    # 归档旧数据
                    archive_writer.writerow(row)
            except (ValueError, IndexError):
                # 跳过格式错误的无效行
                continue
    
    # 删除临时文件
    os.remove(temp_file)

# 用schedule库定时执行（或用系统cron任务更稳定）
import schedule
schedule.every().hour.do(archive_old_data)

while True:
    schedule.run_pending()
    time.sleep(60)

关键注意事项 #

• 原子操作的必要性：os.rename在Linux/Windows下都是原子操作，确保原文件不会出现丢失或损坏的情况。
• 逐行处理效率：40k+条记录逐行处理仅需几毫秒，完全不会占用过多内存或时间。
• 定时任务选择：生产环境建议用系统定时任务（如Linux的cron、Windows的任务计划程序）替代schedule库，避免归档脚本意外崩溃。

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