从两个CSV导入大数据时处理MySQL外键约束问题

方法1：修改INSERT语句，通过EXISTS约束仅插入匹配项 #

直接调整SQL语句，用INSERT ... SELECT替代INSERT ... VALUES，同时通过WHERE EXISTS子句检查当前sampleName是否存在于表A的Name字段中，数据库会自动过滤不匹配的记录，不会触发外键约束错误。

修改后的SQL及Python代码示例：

INSERT INTO B(abc, sampleName, col3, col4, col5)
SELECT %s, %s, %s, %s, %s
WHERE EXISTS (SELECT 1 FROM A WHERE A.Name = %s)

# 假设CSV处理后的X是(abc, sampleName, col3, col4, col5)格式的元组
Q = """
INSERT INTO B(abc, sampleName, col3, col4, col5)
SELECT %s, %s, %s, %s, %s
WHERE EXISTS (SELECT 1 FROM A WHERE A.Name = %s)
"""
# 执行时将sampleName作为匹配条件追加到参数末尾
cursor1.execute(Q, (*X, X[1]))

# 批量插入场景
params = [(*row, row[1]) for row in data_list]
cursor1.executemany(Q, params)

方法2：Python预处理过滤数据（适合大数据量，减少数据库交互） #

如果CSV数据量极大，多次执行带EXISTS的INSERT效率偏低，可以先从表A获取所有有效Name，在Python层面过滤不匹配的行后再批量插入：

1. 查询表A所有有效Name存入集合（集合查找效率远高于列表）：

cursor1.execute("SELECT Name FROM A")
valid_names = {row[0] for row in cursor1.fetchall()}

2. 读取CSV时过滤无效行：

import csv

data_list = []
with open('your_csv_file.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    for row in reader:
        sample_name = row['sampleName']
        if sample_name in valid_names:
            data_tuple = (row['abc'], sample_name, row['col3'], row['col4'], row['col5'])
            data_list.append(data_tuple)

3. 批量插入过滤后的数据：

Q = "INSERT INTO B(abc, sampleName, col3, col4, col5) VALUES(%s, %s, %s, %s, %s)"
cursor1.executemany(Q, data_list)
conn.commit()

额外提示 #

• 方法1依赖数据库处理，方法2减少数据库查询次数，适合超大数据量场景。
• 若Name字段存在大小写敏感问题（如字符集为utf8_bin），可统一转换大小写后匹配，比如sample_name.lower() in {name.lower() for name in valid_names}。

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