如何在Python中不使用Pandas合并两个含共同ID的大CSV文件？

问题说明 #

我有两个CSV文件需要合并：

• client.csv：15GB，1.6亿行，结构如下：

• phone.csv：结构如下：

要求合并后输出CSV或SQLite表，client里的id如果在phone里找不到匹配，phone列留空。之前用Pandas处理时因为文件太大直接爆内存了，求不用Pandas的解决办法。

方案1：原生CSV模块+字典缓存（适合phone.csv不大的情况） #

如果phone.csv体积小（能装到内存里），这是最简单的办法：先把phone的id和对应号码存到字典，再逐行读client.csv，匹配后写入结果。

import csv

# 先加载phone.csv到字典，key是id，value是手机号
phone_map = {}
with open('phone.csv', 'r', encoding='utf-8') as phone_file:
    reader = csv.DictReader(phone_file)
    for row in reader:
        phone_map[row['id']] = row['phone']

# 逐行处理client.csv，匹配手机号后写入结果
with open('client.csv', 'r', encoding='utf-8') as client_file, \
     open('merged_result.csv', 'w', encoding='utf-8', newline='') as result_file:
    
    client_reader = csv.DictReader(client_file)
    # 构造结果表头：client的所有字段 + phone
    output_fields = client_reader.fieldnames + ['phone']
    writer = csv.DictWriter(result_file, fieldnames=output_fields)
    writer.writeheader()
    
    for row in client_reader:
        # 找对应手机号，没有就留空
        row['phone'] = phone_map.get(row['id'], '')
        writer.writerow(row)

方案2：排序后双指针合并（适合两个文件都很大的情况） #

如果phone.csv也大到装不下内存，那就先给两个文件按id排序，再用双指针逐行合并，类似归并排序的思路。

第一步：给两个CSV按id排序（用系统命令更快） #

Linux/macOS直接用sort命令：

# 给client.csv按第一列（id）排序，输出到sorted_client.csv
sort -t',' -k1,1 client.csv > sorted_client.csv

# 给phone.csv按第一列排序，输出到sorted_phone.csv
sort -t',' -k1,1 phone.csv > sorted_phone.csv

Windows可以用Git Bash的sort，或者PowerShell的Sort-Object。

第二步：Python逐行合并排序后的文件 #

import csv

def merge_sorted_csvs(client_path, phone_path, output_path):
    with open(client_path, 'r', encoding='utf-8') as client_f, \
         open(phone_path, 'r', encoding='utf-8') as phone_f, \
         open(output_path, 'w', encoding='utf-8', newline='') as out_f:
        
        client_reader = csv.DictReader(client_f)
        phone_reader = csv.DictReader(phone_f)
        output_fields = client_reader.fieldnames + ['phone']
        writer = csv.DictWriter(out_f, fieldnames=output_fields)
        writer.writeheader()
        
        # 初始化phone的行指针
        current_phone_row = next(phone_reader, None)
        
        for client_row in client_reader:
            client_id = client_row['id']
            # 移动phone指针，直到找到大于等于当前client id的行
            while current_phone_row is not None and current_phone_row['id'] < client_id:
                current_phone_row = next(phone_reader, None)
            
            # 匹配手机号
            if current_phone_row is not None and current_phone_row['id'] == client_id:
                client_row['phone'] = current_phone_row['phone']
                # 移动指针，避免重复匹配同一个id
                current_phone_row = next(phone_reader, None)
            else:
                client_row['phone'] = ''
            
            writer.writerow(client_row)

# 调用合并函数
merge_sorted_csvs('sorted_client.csv', 'sorted_phone.csv', 'merged_result.csv')

方案3：用SQLite处理（最稳妥的大文件方案） #

SQLite天生适合处理大文件，不用把数据全加载到内存，直接导入CSV后用SQL做左连接就行。

方法一：用Python脚本处理 #

import sqlite3
import csv

# 连接SQLite数据库（不存在则自动创建）
conn = sqlite3.connect('merged_data.db')
cursor = conn.cursor()

# 创建client表
cursor.execute('''
CREATE TABLE client (
    id TEXT PRIMARY KEY,
    name TEXT,
    email TEXT,
    contract_number TEXT,
    address_id TEXT
)
''')

# 创建phone表
cursor.execute('''
CREATE TABLE phone (
    id TEXT PRIMARY KEY,
    phone TEXT
)
''')

# 逐行导入client.csv，避免内存溢出
with open('client.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    for row in reader:
        cursor.execute('''
        INSERT INTO client (id, name, email, contract_number, address_id)
        VALUES (?, ?, ?, ?, ?)
        ''', (row['id'], row['name'], row['email'], row['contract_number'], row['address_id']))

# 导入phone.csv
with open('phone.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    for row in reader:
        cursor.execute('''
        INSERT INTO phone (id, phone)
        VALUES (?, ?)
        ''', (row['id'], row['phone']))

conn.commit()

# 执行左连接查询并导出到CSV
cursor.execute('''
SELECT c.*, p.phone
FROM client c
LEFT JOIN phone p ON c.id = p.id
''')

with open('merged_result.csv', 'w', encoding='utf-8', newline='') as f:
    writer = csv.writer(f)
    # 写入表头
    writer.writerow([desc[0] for desc in cursor.description])
    # 逐行写入查询结果
    for row in cursor:
        writer.writerow(row)

conn.close()

方法二：直接用SQLite命令行（效率更高） #

如果熟悉命令行，直接用SQLite的内置命令更快：

# 创建数据库并完成导入、查询、导出
sqlite3 merged_data.db <<EOF
.mode csv
.import client.csv client
.import phone.csv phone
.headers on
.output merged_result.csv
SELECT c.*, p.phone FROM client c LEFT JOIN phone p ON c.id = p.id;
EOF

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