4. 用户实体表构建与唯一ID生成
任务说明
本案例从三类原始数据（注册表、登录日志、支付记录）中抽取用户实体，设计稳定唯一ID生成规则，构建用户节点表，并处理同一用户多标识（手机号、邮箱、设备号）的合并。
核心思路：以手机号作为主标识，邮箱和设备号作为辅助标识。若同一用户在不同数据源中出现不同标识，通过标识交叉引用进行合并，最终为每个用户分配一个稳定的 UID（格式 U_ + 8位哈希）。
环境与依赖
Python 3.8+
pandas
安装：pip install pandas
完整实现
import pandas as pd
import hashlib
# ---------- 1. 模拟多源原始数据 ----------
register = pd.DataFrame({
"phone": ["13800000001", "13800000002", "13800000003"],
"email": ["a@x.com", "b@x.com", None],
"name": ["张三", "李四", "王五"]
})
login_log = pd.DataFrame({
"phone": ["13800000001", "13800000002", "13800000004"],
"device": ["D001", "D002", "D003"]
})
pay_log = pd.DataFrame({
"phone": ["13800000001", "13800000003", "13800000004"],
"email": ["a@x.com", None, "d@x.com"],
"device": ["D001", "D004", "D003"]
})
# ---------- 2. 抽取用户实体（phone 为主键，合并辅助标识） ----------
def extract_entities(*frames):
"""从多源数据中抽取 (phone, email, device) 三元组，去重。"""
records = []
for df in frames:
cols = {c: df[c] if c in df.columns else None for c in ("phone", "email", "device")}
n = len(df)
for i in range(n):
records.append((
cols["phone"].iloc[i] if cols["phone"] is not None else None,
cols["email"].iloc[i] if cols["email"] is not None else None,
cols["device"].iloc[i] if cols["device"] is not None else None,
))
# 去重
return list(dict.fromkeys(records))
entities = extract_entities(register, login_log, pay_log)
# ---------- 3. 标识合并：并查集 ----------
class UnionFind:
def __init__(self):
self.parent = {}
def find(self, x):
self.parent.setdefault(x, x)
while self.parent[x] != x:
self.parent[x] = self.parent[self.parent[x]]
x = self.parent[x]
return x
def union(self, a, b):
ra, rb = self.find(a), self.find(b)
if ra != rb:
self.parent[rb] = ra
uf = UnionFind()
for phone, email, device in entities:
ids = [v for v in (phone, email, device) if v]
for other in ids[1:]:
uf.union(ids[0], other)
# ---------- 4. 生成稳定唯一ID ----------
def make_uid(key: str) -> str:
return "U_" + hashlib.md5(key.encode()).hexdigest()[:8]
# 每个连通分量选一个代表标识（优先 phone），生成 UID
groups = {}
for phone, email, device in entities:
for v in (phone, email, device):
if v:
root = uf.find(v)
groups.setdefault(root, set()).add(v)
# 选代表：优先 phone，否则字典序最小
def pick_rep(ids):
phones = sorted(i for i in ids if i.startswith("1") and i.isdigit())
return phones[0] if phones else sorted(ids)[0]
uid_map = {} # 标识 -> UID
user_rows = []
for root, ids in groups.items():
rep = pick_rep(ids)
uid = make_uid(rep)
for i in ids:
uid_map[i] = uid
user_rows.append({
"uid": uid,
"phone": next((i for i in ids if i.startswith("1") and i.isdigit()), None),
"email": next((i for i in ids if "@" in i), None),
"device": next((i for i in ids if i.startswith("D")), None),
})
user_table = pd.DataFrame(user_rows).sort_values("uid").reset_index(drop=True)
# ---------- 5. 输出与验证 ----------
print("=== 用户节点表 ===")
print(user_table.to_string(index=False))
print("\n=== 标识 -> UID 映射 ===")
for k, v in sorted(uid_map.items()):
print(f" {k:15s} -> {v}")
# 验证：同一用户多标识应映射到同一 UID
assert uid_map["13800000001"] == uid_map["a@x.com"] == uid_map["D001"]
assert uid_map["13800000004"] == uid_map["D003"] == uid_map["d@x.com"]
print("\n[OK] 多标识合并验证通过")
运行方式
python user_entity.py验证方法与通过条件
- 输出用户节点表，每行一个用户，包含
uid / phone / email / device四列。 - 标识映射中，
13800000001、a@x.com、D001应指向同一 UID（同一用户跨源合并）。 13800000004、D003、d@x.com应指向同一 UID。- 脚本末尾
assert全部通过并打印[OK]即判定成功。
关键设计说明
- 稳定ID：UID 由代表标识的 MD5 前8位生成，同一用户无论数据到达顺序如何，UID 不变。
- 多标识合并：使用并查集将同一用户的不同标识（手机号/邮箱/设备号）归入同一连通分量。
- 代表选取：优先手机号作为代表标识，保证 UID 可读且稳定。