高效合并重复联系人并生成Master Contact ID的方案咨询

需求概述 #

需要基于以下三个条件将重复联系人分组，并为每组生成唯一的Master Contact ID：

• 匹配Email
• 匹配Phone&Name（电话号码与姓名组合）
• 匹配Account&Name（账户与姓名组合）

原始数据 #

ContactID   Name           Email               Phone&Name               Account&Name
12345       Bob Smith      Bob@ABC.com         234-243-2432Bob Smith    A1234Bob Smith
42023       Bob Smith      Bob01@ABC.com       234-243-2432Bob Smith    B1234Bob Smith
50203       Bob S.         Bob@ABC.com         234-243-2432Bob S.       Z1234Bob S.
20394       Clara Sakshi   Clara@Sakshi.com    123-123-1234Clara Sakshi Q1231Clara Sakshi
29930       Clara Sakshi   Clara@ABC.com       234-243-2432Clara Sakshi A1234Clara Sakshi
92303       Clara Sakshi   Clara01@Sakshi.com  999-999-1234Clara Sakshi Q1231Clara Sakshi

期望输出 #

Master ContactID   ContactID    Notes (not part of output):
1                  12345        related to 50203 by email match
1                  42023        related to 12345 by name and number match
1                  50203        related to 12345 by email match
2                  20394        related to 92303 by account number and name match
3                  29930        Not related to any other Contacts
2                  92303        related to 20394 by account number and name match

现有方案问题 #

之前通过SQL逆透视联系人表，再应用图遍历技术实现，但性能极差：1000条数据耗时近1小时，数据量增大时运行时间呈指数增长，无法处理25万条联系人数据。

高效解决方案 #

方案一：Python + 并查集（Union-Find）算法 #

并查集是处理连通分量问题的最优算法之一，时间复杂度接近O(n)，完全适配25万条数据的规模。

实现步骤：

1. 加载联系人数据（可从CSV/数据库读取）
2. 初始化并查集，每个ContactID初始为自己的父节点
3. 按三个匹配条件依次合并连通的ContactID：
   • 按Email分组，同一Email下的所有ContactID合并
   • 按Phone&Name分组，同一组合下的所有ContactID合并
   • 按Account&Name分组，同一组合下的所有ContactID合并
4. 为每个连通分量分配唯一的Master Contact ID

代码示例：

class UnionFind:
    def __init__(self, elements):
        self.parent = {elem: elem for elem in elements}
    
    def find(self, x):
        if self.parent[x] != x:
            self.parent[x] = self.find(self.parent[x])  # 路径压缩
        return self.parent[x]
    
    def union(self, x, y):
        x_root = self.find(x)
        y_root = self.find(y)
        if x_root != y_root:
            self.parent[y_root] = x_root

# 模拟加载数据（实际可从CSV/数据库读取）
contacts = [
    (12345, "Bob Smith", "Bob@ABC.com", "234-243-2432Bob Smith", "A1234Bob Smith"),
    (42023, "Bob Smith", "Bob01@ABC.com", "234-243-2432Bob Smith", "B1234Bob Smith"),
    (50203, "Bob S.", "Bob@ABC.com", "234-243-2432Bob S.", "Z1234Bob S."),
    (20394, "Clara Sakshi", "Clara@Sakshi.com", "123-123-1234Clara Sakshi", "Q1231Clara Sakshi"),
    (29930, "Clara Sakshi", "Clara@ABC.com", "234-243-2432Clara Sakshi", "A1234Clara Sakshi"),
    (92303, "Clara Sakshi", "Clara01@Sakshi.com", "999-999-1234Clara Sakshi", "Q1231Clara Sakshi"),
]

# 初始化并查集
contact_ids = [c[0] for c in contacts]
uf = UnionFind(contact_ids)

# 按Email合并
email_groups = {}
for c in contacts:
    email = c[2]
    if email not in email_groups:
        email_groups[email] = []
    email_groups[email].append(c[0])

for group in email_groups.values():
    if len(group) > 1:
        first = group[0]
        for elem in group[1:]:
            uf.union(first, elem)

# 按Phone&Name合并
phone_name_groups = {}
for c in contacts:
    key = c[3]
    if key not in phone_name_groups:
        phone_name_groups[key] = []
    phone_name_groups[key].append(c[0])

for group in phone_name_groups.values():
    if len(group) > 1:
        first = group[0]
        for elem in group[1:]:
            uf.union(first, elem)

# 按Account&Name合并
account_name_groups = {}
for c in contacts:
    key = c[4]
    if key not in account_name_groups:
        account_name_groups[key] = []
    account_name_groups[key].append(c[0])

for group in account_name_groups.values():
    if len(group) > 1:
        first = group[0]
        for elem in group[1:]:
            uf.union(first, elem)

# 生成Master Contact ID
root_map = {}
current_master_id = 1
for cid in contact_ids:
    root = uf.find(cid)
    if root not in root_map:
        root_map[root] = current_master_id
        current_master_id += 1

# 输出结果
print("Master ContactID   ContactID")
for cid in contact_ids:
    print(f"{root_map[uf.find(cid)]:20} {cid}")

方案二：SQL优化方案（适用于数据库端处理） #

如果必须用SQL处理，避免图遍历，改用递归CTE + 分组合并的方式，核心是先按三个条件生成关联对，再通过递归合并连通分量。

实现步骤：

1. 创建临时表存储所有关联关系（同一Email/Phone&Name/Account&Name的ContactID对）
2. 用递归CTE遍历所有连通的ContactID，为每个连通分量标记根节点
3. 基于根节点分配Master Contact ID

代码示例（以MySQL为例）：

-- 1. 创建临时表存储所有关联对
CREATE TEMPORARY TABLE contact_links AS
-- Email关联
SELECT a.ContactID AS cid1, b.ContactID AS cid2
FROM contacts a
JOIN contacts b ON a.Email = b.Email AND a.ContactID < b.ContactID
UNION
-- Phone&Name关联
SELECT a.ContactID AS cid1, b.ContactID AS cid2
FROM contacts a
JOIN contacts b ON a.`Phone&Name` = b.`Phone&Name` AND a.ContactID < b.ContactID
UNION
-- Account&Name关联
SELECT a.ContactID AS cid1, b.ContactID AS cid2
FROM contacts a
JOIN contacts b ON a.`Account&Name` = b.`Account&Name` AND a.ContactID < b.ContactID;

-- 2. 递归CTE合并连通分量
WITH RECURSIVE contact_groups AS (
    SELECT cid1 AS root, cid1 AS cid FROM contact_links
    UNION ALL
    SELECT g.root, l.cid2 AS cid
    FROM contact_groups g
    JOIN contact_links l ON g.cid = l.cid1
    WHERE l.cid2 NOT IN (SELECT cid FROM contact_groups WHERE root = g.root)
    UNION ALL
    SELECT g.root, l.cid1 AS cid
    FROM contact_groups g
    JOIN contact_links l ON g.cid = l.cid2
    WHERE l.cid1 NOT IN (SELECT cid FROM contact_groups WHERE root = g.root)
),
-- 补充孤立的ContactID
all_contacts AS (
    SELECT ContactID AS cid FROM contacts
)
-- 3. 分配Master Contact ID
SELECT 
    DENSE_RANK() OVER (ORDER BY COALESCE(g.root, a.cid)) AS `Master ContactID`,
    a.cid AS `ContactID`
FROM all_contacts a
LEFT JOIN contact_groups g ON a.cid = g.cid
GROUP BY a.cid
ORDER BY `Master ContactID`, `ContactID`;

注意：SQL方案需确保数据库有足够的内存和递归深度设置，对于25万条数据，建议先对三个匹配字段创建索引，大幅提升关联查询速度。

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