账户合并与连通分量
将电子邮件视为 DSU 节点，把共享电子邮件的账户分组，然后收集每个分量中的所有电子邮件以重建合并后的账户。
账户合并与连通分量 是 CoddyKit 上的免费 DSA Interview Prep 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 DSA Interview Prep 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 DSA Interview Prep 课程共包含 4 节课。
问题：账户合并
账户合并问题（LeetCode 721）给定一个账户列表，其中每个账户都是一个字符串列表：第一个元素是账户名称，其余元素是电子邮件地址。如果两个账户至少共享一个电子邮件地址，则它们属于同一个（same）人。请合并属于同一人的所有账户，并返回已排序的电子邮件列表。
这本质上是一个连通分量问题，其中电子邮件是节点，共享账户会将它们连接起来。DSU 是理想工具：对同一账户中的所有电子邮件执行 union 操作，然后按连通分量收集电子邮件。
# Example input
accounts = [
['John', 'john@mail.com', 'john1@mail.com'],
['John', 'john2@mail.com'],
['Mary', 'mary@mail.com'],
['John', 'john1@mail.com', 'john2@mail.com'],
]
# john@mail.com and john1@mail.com are in account[0]
# john1@mail.com and john2@mail.com are in account[3]
# => john@, john1@, john2@ are all the same person
# Expected output:
# ['John', 'john1@mail.com', 'john2@mail.com', 'john@mail.com']
# ['Mary', 'mary@mail.com']
print('Goal: merge accounts sharing any email into one account')将电子邮件映射为整数 ID
DSU 使用整数索引，但我们的节点是电子邮件字符串。我们需要将每个唯一的电子邮件映射到一个整数 ID。此外，我们还需要记住每封电子邮件所属的名称。使用字典 email_to_id 分配递增的 ID，并使用 email_to_name 跟踪与每封电子邮件关联的账户名称。
每封唯一的电子邮件都会获得一个 ID。如果同一（same）封电子邮件出现在多个账户中，它会映射到同一个 ID——对同一账户内电子邮件的 ID 执行 union 操作，就能将它们连接为一个连通分量。与根电子邮件的 ID 关联的名称就是合并后账户的名称。
accounts = [
['John', 'john@mail.com', 'john1@mail.com'],
['John', 'john2@mail.com'],
['Mary', 'mary@mail.com'],
['John', 'john1@mail.com', 'john2@mail.com'],
]
email_to_id = {}
email_to_name = {}
next_id = [0]
for account in accounts:
name = account[0]
for email in account[1:]:
if email not in email_to_id:
email_to_id[email] = next_id[0]
next_id[0] += 1
email_to_name[email] = name
print('Total unique emails:', len(email_to_id))
for email, eid in email_to_id.items():
print(f' {email} => id {eid} (owner: {email_to_name[email]})')合并每个账户中的电子邮件
对于每个账户，我们会对其中列出的所有电子邮件的 ID 执行 union 操作。我们选择账户中的第一封电子邮件作为代表，并将其他每封电子邮件的 ID 与它执行 union 操作。这样就能将账户中的所有电子邮件连接到同一个连通分量。
处理完所有账户后，共同出现过的电子邮件（无论是直接出现，还是通过跨账户共享电子邮件传递连接的）都会共享同一个 DSU 根节点。这是跨多个账户传播连通性的关键步骤。
class DSU:
def __init__(self, n):
self.parent = list(range(n))
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):
self.parent[self.find(x)] = self.find(y)
# After building email_to_id (from previous step)
# email_to_id = {'john@mail.com':0, 'john1@mail.com':1,
# 'john2@mail.com':2, 'mary@mail.com':3}
dsu = DSU(5) # 4 unique emails
# For account ['John', 'john@mail.com', 'john1@mail.com']:
dsu.union(0, 1) # john@ and john1@ share account => same component
# For account ['John', 'john1@mail.com', 'john2@mail.com']:
dsu.union(1, 2) # john1@ and john2@ share account => same component
# Now 0,1,2 all share a root; 3 (mary) is separate
print('find(0)==find(2)?', dsu.find(0) == dsu.find(2)) # True
print('find(0)==find(3)?', dsu.find(0) == dsu.find(3)) # False收集每个连通分量中的电子邮件
完成所有 union 操作后，我们遍历每封电子邮件，执行 find 找到其 DSU 根节点，并使用一个列表字典按该根节点对电子邮件进行分组。根节点 ID 作为键。最后，对于每个分组，我们获取账户名称，对电子邮件列表执行 sort，并将名称放在列表开头。
对电子邮件执行 sort 是题目的要求——在合并后的账户中，电子邮件必须按字典序排列。名称可以从分组中的任意一封电子邮件获取，因为一个连通分量中的所有电子邮件都属于同一个（same）人。
from collections import defaultdict
# After DSU unions, group by root
def collect_components(email_to_id, email_to_name, dsu):
root_to_emails = defaultdict(list)
for email, eid in email_to_id.items():
root = dsu.find(eid)
root_to_emails[root].append(email)
result = []
for root, emails in root_to_emails.items():
# Find the name from any email in this group
name = email_to_name[emails[0]]
result.append([name] + sorted(emails))
return result
# Mock data for illustration
email_to_id = {'john@m.com':0,'john1@m.com':1,'john2@m.com':2,'mary@m.com':3}
email_to_name = {e:'John' for e in list(email_to_id)[:3]}
email_to_name['mary@m.com'] = 'Mary'
class DSU:
def __init__(self,n): self.p=list(range(n))
def find(self,x): self.p[x]=self.p[self.p[x]] if self.p[x]!=x else x; return self.p[x] if self.p[x]==x else self.find(self.p[x])
def union(self,x,y): self.p[self.find(x)]=self.find(y)
dsu=DSU(4); dsu.union(0,1); dsu.union(1,2)
for row in collect_components(email_to_id, email_to_name, dsu):
print(row)账户合并完整解法
下面是结合全部三个步骤的完整解法：构建电子邮件到 ID 的映射，对每个账户中的电子邮件执行 union 操作，并按 DSU 根节点收集分组后的电子邮件。总体时间复杂度为O(n × m × alpha(n × m))，其中 n 是账户数量，m 是每个账户中的最大电子邮件数量，实际上为 O(n × m)。
空间复杂度为 O(n × m)，用于存储电子邮件映射和 DSU 数组。该解法能够正确处理传递式合并：如果账户 A 与账户 B 共享电子邮件 X，而账户 B 与账户 C 共享电子邮件 Y，那么 A、B 和 C 都会合并到同一个分组中。
from collections import defaultdict
def accounts_merge(accounts):
email_to_id = {}
email_to_name = {}
eid = 0
for account in accounts:
name = account[0]
for email in account[1:]:
if email not in email_to_id:
email_to_id[email] = eid
eid += 1
email_to_name[email] = name
parent = list(range(eid))
def find(x):
if parent[x] != x:
parent[x] = find(parent[x])
return parent[x]
def union(x, y):
parent[find(x)] = find(y)
for account in accounts:
first_id = email_to_id[account[1]]
for email in account[2:]:
union(first_id, email_to_id[email])
root_to_emails = defaultdict(list)
for email, i in email_to_id.items():
root_to_emails[find(i)].append(email)
return [[email_to_name[emails[0]]] + sorted(emails)
for emails in root_to_emails.values()]
accounts = [['John','a@m.com','b@m.com'],['John','c@m.com'],
['Mary','d@m.com'],['John','b@m.com','c@m.com']]
for row in accounts_merge(accounts):
print(row)账户合并中的 BFS/DFS 替代方案
另一种方法是构建电子邮件到账户的关联图：电子邮件作为节点，属于同一账户的电子邮件之间建立边。然后使用 BFS/DFS 找出每个连通分量。虽然这种方法是正确的，但需要显式构建图，并从每个尚未访问的电子邮件开始运行 BFS；与 DSU 相比，代码更多，也更难推理。
DSU 更简洁，因为并查集结构可以自然地表示分量归属，无需显式的邻接表。只有在需要重建两个账户之间共享电子邮件的实际路径或链路时，BFS 才更适合。
# BFS alternative (for comparison)
from collections import defaultdict, deque
def accounts_merge_bfs(accounts):
email_to_accounts = defaultdict(set)
for i, account in enumerate(accounts):
for email in account[1:]:
email_to_accounts[email].add(i)
visited_accounts = set()
result = []
for i, account in enumerate(accounts):
if i in visited_accounts:
continue
queue = deque([i])
emails_in_group = set()
while queue:
acc_idx = queue.popleft()
if acc_idx in visited_accounts:
continue
visited_accounts.add(acc_idx)
for email in accounts[acc_idx][1:]:
emails_in_group.add(email)
for j in email_to_accounts[email]:
queue.append(j)
result.append([account[0]] + sorted(emails_in_group))
return result
accounts = [['John','a@m.com','b@m.com'],['John','b@m.com','c@m.com'],['Mary','d@m.com']]
for row in accounts_merge_bfs(accounts):
print(row)泛化：图的连通分量
账户合并模式可以泛化到任何带标签的连通分量问题：您有一组元素，其中一些元素被声明为等价（连通），您希望将所有传递等价的元素归为一组。示例包括聚类问题、社交网络中的好友群组，以及重复记录检测。
通用算法始终是：(1) 为每个元素分配一个整数 ID；(2) 对声明为等价的元素执行 ID 的 union；(3) 根据 DSU 根节点对元素进行分组。DSU 本质上是处理等价关系的分组引擎。
# Generalised grouping template
def group_equivalents(items, equivalences):
item_to_id = {item: i for i, item in enumerate(items)}
n = len(items)
parent = list(range(n))
def find(x):
if parent[x] != x:
parent[x] = find(parent[x])
return parent[x]
def union(x, y):
parent[find(x)] = find(y)
for a, b in equivalences:
if a in item_to_id and b in item_to_id:
union(item_to_id[a], item_to_id[b])
groups = {}
for item in items:
root = find(item_to_id[item])
groups.setdefault(root, []).append(item)
return list(groups.values())
# Example: merging duplicate customer records
customers = ['Alice-NY','Alice-LA','Bob','Alice-TX','Carol']
links = [('Alice-NY','Alice-LA'),('Alice-LA','Alice-TX')]
print(group_equivalents(customers, links))处理边界情况
账户合并中的重要边界情况：
- 只有一个电子邮件的账户：只有一个电子邮件的账户会形成自己的分量，除非另一个账户共享该电子邮件。
- 姓名相同但不是同一个人：两个账户中出现“John”并不意味着它们属于同一个人——只有共享电子邮件才会合并账户。姓名是按电子邮件存储的，而不是按分量存储的。
- 空账户：没有电子邮件的账户应跳过，以避免索引错误。
始终确认您的解决方案不会仅仅因为账户共享姓名，就错误地合并本不应合并的账户。DSU 连接完全由共享的电子邮件地址决定。
# Edge case: two Johns with no shared email => separate output
accounts = [
['John', 'john_a@m.com'],
['John', 'john_b@m.com'], # different email => different component
['Mary'], # no emails => skip
]
def accounts_merge_safe(accounts):
email_to_id = {}; email_to_name = {}; eid = 0
for account in accounts:
name = account[0]
for email in account[1:]:
if email not in email_to_id:
email_to_id[email] = eid; eid += 1
email_to_name[email] = name
parent = list(range(eid))
def find(x):
while parent[x]!=x: parent[x]=parent[parent[x]]; x=parent[x]
return x
def union(x,y): parent[find(x)]=find(y)
for account in accounts:
if len(account) < 2: continue # skip no-email accounts
first = email_to_id[account[1]]
for email in account[2:]:
union(first, email_to_id[email])
from collections import defaultdict
groups = defaultdict(list)
for email, i in email_to_id.items():
groups[find(i)].append(email)
return [[email_to_name[e[0]]] + sorted(e) for e in groups.values()]
for row in accounts_merge_safe(accounts):
print(row)图中的连通分量数量
一个相关问题（LeetCode 323）要求计算无向图中的连通分量数量。这比账户合并更简单：用 n 个节点初始化 DSU，使用 union 处理所有边，然后统计不同的根节点。
统计分量最简洁的方法是维护一个从 n 开始的 count 变量，每当一次成功的 union 合并两个不同的分量时，就将它减一。另一种方法是在最后统计满足 find(i) == i 的节点 i 的数量。
def count_components(n, edges):
parent = list(range(n))
def find(x):
if parent[x] != x:
parent[x] = find(parent[x])
return parent[x]
count = n
for u, v in edges:
pu, pv = find(u), find(v)
if pu != pv:
parent[pu] = pv
count -= 1
return count
print(count_components(5, [[0,1],[1,2],[3,4]])) # 2: {0,1,2} and {3,4}
print(count_components(5, [[0,1],[1,2],[2,3],[3,4]])) # 1: all connected
print(count_components(5, [])) # 5: no edges, all isolated最小分量和最大分量
有了支持大小跟踪的 DSU 后，您就可以回答“最大的连通分量有多大？”或“恰好有 3 个节点的分量有多少个？”之类的查询：扫描根节点对应的大小数组即可在 O(n) 时间内完成。
这类查询会出现在“寻找网格中最大的连通岛屿”或“识别最小网络分区”等问题中。所有 union 完成后，扫描满足 find(i) == i 的节点 i（这些节点就是根节点），并检查它们的大小。
class DSU:
def __init__(self, n):
self.parent = list(range(n))
self.size = [1] * n
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):
px, py = self.find(x), self.find(y)
if px == py: return
if self.size[px] < self.size[py]: px, py = py, px
self.parent[py] = px
self.size[px] += self.size[py]
def component_stats(n, edges):
dsu = DSU(n)
for u, v in edges:
dsu.union(u, v)
sizes = [dsu.size[i] for i in range(n) if dsu.find(i) == i]
print('Component sizes:', sizes)
print('Largest component:', max(sizes))
print('Smallest component:', min(sizes))
print('Number of components:', len(sizes))
component_stats(8, [(0,1),(1,2),(3,4),(5,6),(6,7)])DSU 问题的面试技巧
当您遇到涉及合并群组、连通性查询或寻找多余边的问题时，应立即想到 DSU。在面试中，请提到两种优化（路径压缩 + 按秩或按大小合并），以展示您对相关知识的深入理解，即使在给定约束下，较简单的朴素 DSU 也能通过。
需要避免的常见错误包括：忘记处理两个端点已经连通的情况（此时 union 不执行任何操作）、错误混用从 0 开始索引和从 1 开始索引，以及没有为账户合并的输出排序（题目要求电子邮件列表有序）。编写代码前，请始终确认输入约束。
# Interview checklist for DSU problems
checklist = [
'1. Identify: is this a grouping/connectivity/cycle problem?',
'2. Map problem entities to integer node IDs if needed',
'3. Implement DSU with path compression + union by rank/size',
'4. Process all relationships (edges/pairs) with union()',
'5. Answer queries using find() and size/count tracking',
'6. Handle edge cases: already connected, single nodes, no edges',
'7. Check output format: sorted? 1-indexed? Name included?',
'8. State time complexity: O(n * alpha(n)) ~ O(n)',
]
for item in checklist:
print(item)快速检查
测试您对本课数据结构与算法——编码面试准备相关概念的理解。
课程回顾
本课中您学到了：账户合并是一个连通分量问题，其中电子邮件是节点，账户负责连接电子邮件；DSU 通过将电子邮件映射为整数 ID、合并每个账户中的 ID，再按根节点分组来解决该问题；以及相同的 DSU 分组模板适用于任何等价类或聚类问题。接下来我们将转向位运算，从基本的 AND、OR、XOR、NOT 和移位运算符开始。
常见问题解答
「账户合并与连通分量」课时是免费的吗？
是的 — 「账户合并与连通分量」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 DSA Interview Prep 课程的其余内容，请升级到 CoddyKit PRO。 DSA Interview Prep 课程共包含 4 节课。
「账户合并与连通分量」这节课中我会学到什么？
将电子邮件视为 DSU 节点，把共享电子邮件的账户分组，然后收集每个分量中的所有电子邮件以重建合并后的账户。 你通过在浏览器中直接运行的动手代码来练习 DSA Interview Prep，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 DSA Interview Prep 需要有经验吗？
无需任何先前经验。CoddyKit 上的 DSA Interview Prep 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。
「账户合并与连通分量」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 DSA Interview Prep 课中编写并运行代码吗？
能。每节 DSA Interview Prep 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- 带路径压缩的 DSU
- 按秩合并与反阿克曼函数界
- 冗余连接与环检测
- 账户合并与连通分量