如何高效计算两个大型CSV文件的差异并导出新增行？

你的问题太常见了——直接用readlines()把4GB的CSV全塞进内存，别说普通机器，就算是高配服务器也容易扛不住。咱们得换思路，逐行处理+高效查询才是大文件处理的正确姿势，下面给你几个实用方案：

方案1：哈希集合逐行处理（内存友好，速度快） #

核心思路是：不存整行文本，只存每行的哈希值（固定长度，内存占用骤降），然后逐行读取第二个文件，检查哈希是否不在第一个文件的哈希集合里，不在就是新增行。

import hashlib

def get_line_hash(line):
    # 用MD5生成固定长度的哈希值，也可以用更轻量的sha1
    return hashlib.md5(line.encode()).hexdigest()

input_file1 = "data.csv"
input_file2 = "data_1.csv"
output_path = "out.csv"

# 先批量读取第一个文件的行哈希，存到集合（查询是O(1)）
file1_hashes = set()
with open(input_file1, 'r', encoding='utf-8') as t1:
    for line in t1:
        file1_hashes.add(get_line_hash(line))

# 逐行处理第二个文件，写入新增行
with open(input_file2, 'r', encoding='utf-8') as t2, open(output_path, 'w', encoding='utf-8') as outFile:
    # 先写入表头（如果两个文件表头一致，直接保留）
    header = t2.readline()
    outFile.write(header)
    
    # 遍历剩余行，检查哈希
    for line in t2:
        line_hash = get_line_hash(line)
        if line_hash not in file1_hashes:
            outFile.write(line)

为什么这个方案高效？ #

• 内存占用：只存哈希值（每个MD5是32个字符），就算第一个文件有1亿行，也只占约3.2GB（比存整行文本小太多）
• 查询速度：集合的in操作是O(1)，遍历第二个文件的速度几乎和读文件一样快

方案2：SQLite数据库辅助（极端大文件，内存占用几乎为0） #

如果第一个文件大到连哈希集合都存不下，那就把哈希值放到磁盘数据库里，用SQL查询代替集合查询，内存占用几乎可以忽略。

import hashlib
import sqlite3

def get_line_hash(line):
    return hashlib.md5(line.encode()).hexdigest()

input_file1 = "data.csv"
input_file2 = "data_1.csv"
output_path = "out.csv"

# 创建临时SQLite数据库（也可以存到磁盘文件，比如'file_hashes.db'）
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
cursor.execute('CREATE TABLE IF NOT EXISTS hashes (hash TEXT PRIMARY KEY)')

# 批量插入第一个文件的哈希值
with open(input_file1, 'r', encoding='utf-8') as t1:
    for line in t1:
        h = get_line_hash(line)
        # 用INSERT OR IGNORE避免重复行的哈希重复存储
        cursor.execute('INSERT OR IGNORE INTO hashes VALUES (?)', (h,))
conn.commit()

# 处理第二个文件，查询数据库判断是否为新增行
with open(input_file2, 'r', encoding='utf-8') as t2, open(output_path, 'w', encoding='utf-8') as outFile:
    header = t2.readline()
    outFile.write(header)
    
    for line in t2:
        h = get_line_hash(line)
        cursor.execute('SELECT 1 FROM hashes WHERE hash = ?', (h,))
        # 没查到就是新增行
        if not cursor.fetchone():
            outFile.write(line)

conn.close()

适合场景： #

• 单个文件超过内存容量（比如8GB文件，机器只有4GB内存）
• 不需要重复运行，磁盘IO的开销可以接受

额外优化：正确处理CSV格式（避免多行字段错误） #

如果你的CSV里有包含换行符的字段（比如备注、描述），直接逐行读会把一个字段拆成多行，导致匹配错误。这时候一定要用Python的csv模块来解析：

import csv
import hashlib

def get_row_hash(row):
    # 把CSV行转换成字符串再哈希，确保相同的行哈希一致
    return hashlib.md5(','.join(row).encode()).hexdigest()

input_file1 = "data.csv"
input_file2 = "data_1.csv"
output_path = "out.csv"

file1_hashes = set()

# 用csv.reader正确解析每行
with open(input_file1, 'r', newline='', encoding='utf-8') as t1:
    reader = csv.reader(t1)
    for row in reader:
        file1_hashes.add(get_row_hash(row))

# 处理第二个文件，用csv.writer保证输出格式正确
with open(input_file2, 'r', newline='', encoding='utf-8') as t2, open(output_path, 'w', newline='', encoding='utf-8') as outFile:
    reader = csv.reader(t2)
    writer = csv.writer(outFile)
    
    # 写入表头
    header = next(reader)
    writer.writerow(header)
    
    for row in reader:
        row_hash = get_row_hash(row)
        if row_hash not in file1_hashes:
            writer.writerow(row)

关键说明： #

• newline=''是CSV模块的要求，避免换行符处理错误
• csv.reader会自动处理包含换行符的字段，保证每行都是一个完整的CSV记录

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