核心思路 #

因为两个数据源的格式和顺序都不一致，直接逐行比对肯定行不通。核心思路是先把两边的数据转换成统一的、可索引的结构——用email + key这个唯一组合作为主键，把数据都整理成「主键: 对应数值」的字典，之后再从三个维度对比差异。

具体实现步骤 #

1. 处理超大CSV文件（内存友好版） #

超大CSV不能一次性全加载到内存，用逐行读取的方式更稳妥：

• 每行分割后提取email、key、value三个字段
• 用元组(email, key)作为字典的键，存储对应的数值（如果遇到重复主键，根据业务需求选择覆盖或记录冲突）

示例Python代码：

import csv

csv_data = {}
with open('large_dataset.csv', 'r', encoding='utf-8') as f:
    reader = csv.reader(f)
    # 如果CSV有表头，先跳过表头行
    # next(reader)
    for row_num, row in enumerate(reader, 1):
        if len(row) != 3:
            print(f"⚠️ 第{row_num}行格式无效，跳过：{row}")
            continue
        email, key, val = row
        # 这里如果有重复主键，可以选择覆盖或者抛出警告
        csv_data[(email, key)] = val.strip()

2. 处理数据库查询结果 #

从数据库拉取数据时，同样提取email、key、value字段，构建和CSV完全一致的字典结构：

# 以MySQL为例，其他数据库替换对应驱动即可
import mysql.connector

db_data = {}
conn = mysql.connector.connect(
    host="your_host",
    user="your_user",
    password="your_pwd",
    database="your_db"
)
cur = conn.cursor()
# 根据实际表结构调整查询语句
cur.execute("SELECT email, `key`, value FROM your_table")
for email, key, val in cur.fetchall():
    # 统一转成字符串，避免类型差异导致的比对错误
    db_data[(email, key)] = str(val).strip()
cur.close()
conn.close()

3. 对比并输出差异 #

现在两个结构一致的字典在手，可以从三个维度排查差异：

• CSV独有条目
• 数据库独有条目
• 主键相同但数值不同的条目

示例代码：

# 1. 找出CSV中有但数据库没有的条目
csv_only_entries = [k for k in csv_data if k not in db_data]
if csv_only_entries:
    print("\n=== CSV中存在但数据库缺失的条目 ===")
    for key in csv_only_entries:
        print(f"{key[0]}, {key[1]}, {csv_data[key]}")

# 2. 找出数据库中有但CSV没有的条目
db_only_entries = [k for k in db_data if k not in csv_data]
if db_only_entries:
    print("\n=== 数据库中存在但CSV缺失的条目 ===")
    for key in db_only_entries:
        print(f"{key[0]}, {key[1]}, {db_data[key]}")

# 3. 找出主键相同但数值不一致的条目
value_mismatches = []
for key in csv_data:
    if key in db_data and csv_data[key] != db_data[key]:
        value_mismatches.append((key, csv_data[key], db_data[key]))
if value_mismatches:
    print("\n=== 主键匹配但数值不一致的条目 ===")
    for key, csv_val, db_val in value_mismatches:
        print(f"主键: {key[0]}, {key[1]} | CSV值: {csv_val} | 数据库值: {db_val}")

针对超大数据量的优化建议 #

• 如果CSV文件大到离谱，可以用pandas的分块读取功能，避免内存溢出：

import pandas as pd
chunk_size = 20000  # 每次读取2万行
for chunk in pd.read_csv('large_dataset.csv', chunksize=chunk_size, names=['email', 'key', 'value']):
    for _, row in chunk.iterrows():
        csv_data[(row['email'], row['key'])] = str(row['value']).strip()

• 数据量极大时，把CSV导入数据库临时表，用SQL直接比对效率更高，比如：

-- 找出数值不一致的条目
SELECT t1.email, t1.`key`, t1.value AS csv_val, t2.value AS db_val
FROM temp_csv_table t1
JOIN your_table t2 ON t1.email = t2.email AND t1.`key` = t2.`key`
WHERE t1.value != t2.value;

• 建议把差异输出到日志文件，而不是只打印在控制台，方便后续排查：

with open('data_diff_log.txt', 'w', encoding='utf-8') as log_file:
    for key in csv_only_entries:
        log_file.write(f"CSV_ONLY: {key[0]}, {key[1]}, {csv_data[key]}\n")
    # 其他差异同理写入

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