如何高效加载增量CSV文件？无时间戳场景的实现方案

1. 无需重复读取整个文件的高效增量加载方法 #

最直接高效的方案是记录文件读取偏移量，仅读取新增的字节内容，核心前提是CSV文件为追加写入模式（新增内容始终在文件末尾）：

• 操作步骤：
  1. 记录上次读取结束时的字节偏移量
  2. 每次加载前检查文件当前大小，若大于记录的偏移量，说明存在新增内容
  3. 使用文件操作的seek()方法直接跳转到偏移量位置，读取后续内容
  4. 更新偏移量为当前文件大小

Python示例代码：

import os
import csv
import time

def load_incremental_csv(file_path, last_offset):
    current_size = os.path.getsize(file_path)
    if current_size <= last_offset:
        return last_offset, []
    
    new_rows = []
    with open(file_path, "r", encoding="utf-8") as f:
        f.seek(last_offset)
        reader = csv.reader(f)
        for row in reader:
            new_rows.append(row)
    return current_size, new_rows

# 初始化偏移量为文件当前大小（首次加载时读取全部内容）
last_offset = os.path.getsize("data.csv")

# 模拟定时加载（每5分钟执行一次）
while True:
    last_offset, rows = load_incremental_csv("data.csv", last_offset)
    if rows:
        # 此处替换为实际数据处理逻辑（如写入数据库）
        print(f"新增{len(rows)}条数据")
    time.sleep(300)

2. 无时间戳时的增量加载方案 #

当CSV文件没有时间戳字段时，可通过以下几种方式实现增量加载，均需依赖文件为追加写入模式：

方式一：结合文件修改时间与偏移量 #

• 同时记录上次加载时的文件修改时间和偏移量
• 每次检查文件修改时间是否晚于记录的时间，若是则执行偏移量读取逻辑
• 优点：避免无意义的文件读取操作，减少系统资源消耗

方式二：监听文件系统写入事件 #

利用系统级文件监控工具，实时感知文件的写入操作，触发增量读取：
Python示例（使用watchdog库）：

import os
import csv
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class CSVIncrementalLoader(FileSystemEventHandler):
    def __init__(self, target_file):
        self.target_file = target_file
        self.last_offset = os.path.getsize(target_file)

    def on_modified(self, event):
        if event.src_path == self.target_file:
            # 避免文件修改过程中的临时触发，等待写入完成
            time.sleep(0.5)
            current_size = os.path.getsize(self.target_file)
            if current_size > self.last_offset:
                with open(self.target_file, "r", encoding="utf-8") as f:
                    f.seek(self.last_offset)
                    reader = csv.reader(f)
                    new_rows = [row for row in reader]
                    print(f"新增{len(new_rows)}条数据")
                    # 此处替换为实际数据处理逻辑
                self.last_offset = current_size

if __name__ == "__main__":
    file_path = "data.csv"
    event_handler = CSVIncrementalLoader(file_path)
    observer = Observer()
    observer.schedule(event_handler, path=os.path.dirname(file_path), recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

方式三：记录已读取行号（仅适用于固定行格式） #

• 每次读取后记录已处理的总行数，下次加载时从该行数开始读取
• 注意：若文件中间行被修改或删除，此方法会导致数据重复或丢失，仅适合严格追加的场景

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