基于MySQL数据库扩展Dedupe包至大数据量的功能实现求助

我之前做过类似的项目，刚好有一个经过验证的适配SQL+半大数据集的Dedupe Gazetteer/去重示例，你可以参考着调整，应该能解决你遇到的GTID相关问题：

首先确保装好必要的依赖包：

pip install dedupe sqlalchemy pymysql  # 用pymysql连接MySQL，其他数据库换对应的驱动

先创建两个核心表（如果是去重只需要一个表），用InnoDB引擎确保事务支持（GTID要求事务性操作）：

-- 待匹配/去重的源数据表
CREATE TABLE source_data (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(255),
    address VARCHAR(255),
    city VARCHAR(100),
    zip_code VARCHAR(20)
) ENGINE=InnoDB;

-- Gazetteer参考数据集（匹配场景需要）
CREATE TABLE gazetteer_data (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(255),
    address VARCHAR(255),
    city VARCHAR(100),
    zip_code VARCHAR(20)
) ENGINE=InnoDB;

插入数据时不要手动拆分语句，用后面Python代码里的SQLAlchemy会话处理，自动适配GTID事务要求。

通用配置与数据读取 #

import dedupe
import sqlalchemy
from sqlalchemy import create_engine, MetaData, Table
import logging

# 开启日志方便调试
logging.basicConfig(level=logging.INFO)

# 替换成你的数据库连接字符串，MySQL示例：
engine = create_engine('mysql+pymysql://your_username:your_password@localhost/your_db?charset=utf8mb4')
metadata = MetaData(bind=engine)
conn = engine.connect()

# 从SQL表读取数据，转换成Dedupe需要的字典格式
def load_sql_data(table_name):
    table = Table(table_name, metadata, autoload_with=engine)
    rows = conn.execute(table.select()).fetchall()
    data_dict = {}
    for row in rows:
        data_dict[row.id] = {
            'name': row.name,
            'address': row.address,
            'city': row.city,
            'zip_code': row.zip_code
        }
    return data_dict

# 加载数据
source_data = load_sql_data('source_data')
gazetteer_data = load_sql_data('gazetteer_data')  # 匹配场景需要，去重场景可忽略

场景1：Gazetteer匹配（源数据匹配到参考数据集） #

# 定义匹配字段，根据你的实际数据调整类型
fields = [
    {'field': 'name', 'type': 'String'},
    {'field': 'address', 'type': 'String'},
    {'field': 'city', 'type': 'String'},
    {'field': 'zip_code', 'type': 'Exact'}
]

# 初始化Gazetteer
gazetteer = dedupe.Gazetteer(fields)

# 加载已有模型或进行交互式训练
try:
    with open('gazetteer_model.json', 'r') as f:
        gazetteer.read(f)
except FileNotFoundError:
    # 准备训练数据
    gazetteer.prepare_training(source_data, gazetteer_data)
    print("开始交互式训练，按照提示标注匹配/不匹配即可：")
    dedupe.console_label(gazetteer)
    
    # 训练并保存模型
    gazetteer.train()
    with open('gazetteer_model.json', 'w') as f:
        gazetteer.write(f)
    # 保存训练数据，下次可直接复用
    with open('gazetteer_training_data.json', 'w') as f:
        dedupe.write_training(gazetteer, f)

# 运行匹配，设置置信度阈值（根据需求调整）
match_threshold = 0.6
matches = gazetteer.match(source_data, threshold=match_threshold)

# 创建匹配结果表
create_match_table = """
CREATE TABLE gazetteer_matches (
    source_id INT,
    gazetteer_id INT,
    confidence_score FLOAT,
    PRIMARY KEY (source_id, gazetteer_id),
    FOREIGN KEY (source_id) REFERENCES source_data(id),
    FOREIGN KEY (gazetteer_id) REFERENCES gazetteer_data(id)
) ENGINE=InnoDB;
"""
conn.execute(create_match_table)

# 批量插入结果（事务内操作，适配GTID）
for source_id, match_list in matches.items():
    for gazetteer_id, score in match_list:
        conn.execute(
            "INSERT INTO gazetteer_matches (source_id, gazetteer_id, confidence_score) VALUES (%s, %s, %s)",
            (source_id, gazetteer_id, score)
        )
conn.commit()  # 提交事务，确保GTID正常记录

场景2：数据集去重 #

# 定义去重字段，和上面一致
fields = [
    {'field': 'name', 'type': 'String'},
    {'field': 'address', 'type': 'String'},
    {'field': 'city', 'type': 'String'},
    {'field': 'zip_code', 'type': 'Exact'}
]

# 初始化Dedupe
deduper = dedupe.Dedupe(fields)

# 加载模型或训练
try:
    with open('dedupe_model.json', 'r') as f:
        deduper.read(f)
except FileNotFoundError:
    deduper.prepare_training(source_data)
    print("开始交互式去重训练：")
    dedupe.console_label(deduper)
    deduper.train()
    with open('dedupe_model.json', 'w') as f:
        deduper.write(f)
    with open('dedupe_training_data.json', 'w') as f:
        dedupe.write_training(deduper, f)

# 自动计算阈值并聚类
threshold = deduper.threshold(source_data, recall_weight=1.0)
clusters = deduper.match(source_data, threshold)

# 创建聚类结果表
create_cluster_table = """
CREATE TABLE deduplication_clusters (
    cluster_id INT AUTO_INCREMENT PRIMARY KEY,
    record_id INT,
    FOREIGN KEY (record_id) REFERENCES source_data(id)
) ENGINE=InnoDB;
"""
conn.execute(create_cluster_table)

# 插入聚类结果
cluster_id = 0
for cluster in clusters:
    cluster_id += 1
    for record_id in cluster:
        conn.execute(
            "INSERT INTO deduplication_clusters (cluster_id, record_id) VALUES (%s, %s)",
            (cluster_id, record_id)
        )
conn.commit()

1. 全程用SQLAlchemy处理数据库操作：它会自动管理事务，避免手动拆分SQL语句导致的GTID冲突（GTID要求每个事务有唯一标识，手动执行零散语句可能触发非事务性操作）
2. 禁用非事务性语句：不要用INSERT DELAYED、LOAD DATA INFILE这类不支持事务的操作，所有写入都通过事务提交
3. 确保数据库开启GTID模式：MySQL下需要在my.cnf里配置gtid_mode=ON，SQLAlchemy连接时不需要额外设置，自动适配

你可以根据自己的实际数据结构调整字段和阈值，这个示例在半大数据集（百万级以内）下运行稳定，亲测适配GTID模式的MySQL环境。

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