Python多列聚合：跨用户ID/邮箱统计总播放量与总销售额

规则说明 #

• 涉及两张业务表：
  • USER_PLAYS：存储行号ROW、用户IDUSER_ID、邮箱EMAIL、视频播放量VIDEO_PLAYS，存在两类身份映射场景：用户更换邮箱但保留同一USER_ID、使用已有邮箱注册新USER_ID
  • Sales：存储净销售额NET_SALE、邮箱EMAIL
• 自然人归并逻辑：
  1. 相同USER_ID的记录归属同一用户
  2. 相同EMAIL的记录归属同一用户
  3. 关联关系可传递，所有连通的记录归为同一自然人，为其分配全局唯一UNIQUE_ID
  4. 最终聚合每个UNIQUE_ID对应的总视频播放量、总净销售额

样例数据 #

USER_PLAYS表 #

Sales表 #

期望输出 #

实现方案 #

该需求本质是图连通分量计算问题：把USER_ID、EMAIL都视作图的节点，同一条USER_PLAYS记录内的USER_ID和EMAIL之间建立连边，同一个连通分量内的所有节点对应同一个自然人。
通用实现步骤：

1. 抽取所有USER_ID、EMAIL节点，初始化每个节点的所属集合
2. 使用并查集（Union-Find）算法遍历所有关联边，将连通的节点合并到同一集合
3. 为每个连通集合分配全局唯一UNIQUE_ID
4. 将USER_PLAYS表的播放量按UNIQUE_ID聚合，将Sales表的销售额通过EMAIL关联到对应UNIQUE_ID后聚合
5. 关联两份聚合结果得到最终输出

数据量级较大时，建议用Python/Scala等编程语言实现并查集完成身份打通，性能优于SQL递归，逻辑也更简洁。

以下为Spark SQL/BigQuery 递归CTE实现参考：

-- 抽取所有USER_ID与EMAIL的关联边
WITH edges AS (
  SELECT DISTINCT
    CAST(USER_ID AS STRING) AS node1,
    EMAIL AS node2
  FROM USER_PLAYS
  WHERE USER_ID IS NOT NULL AND EMAIL IS NOT NULL
),
-- 递归遍历所有连通节点
recursive_union AS (
  SELECT node1, node2, ARRAY[node1, node2] AS visited
  FROM edges
  UNION ALL
  SELECT
    r.node1,
    CASE WHEN e.node1 = r.node2 THEN e.node2 ELSE e.node1 END AS node2,
    ARRAY_CONCAT(r.visited, [CASE WHEN e.node1 = r.node2 THEN e.node2 ELSE e.node1 END]) AS visited
  FROM recursive_union r
  JOIN edges e
    ON (r.node2 = e.node1 OR r.node1 = e.node2)
    AND NOT ARRAY_INCLUDES(r.visited, CASE WHEN e.node1 = r.node2 THEN e.node2 ELSE e.node1 END)
),
-- 为每个连通分量分配UNIQUE_ID（取分量内最小节点值作为ID）
node_mapping AS (
  SELECT DISTINCT node, MIN(root_id) AS UNIQUE_ID
  FROM (
    SELECT node1 AS node, MIN(LEAST(node1, node2)) OVER(PARTITION BY node1) AS root_id FROM recursive_union
    UNION ALL
    SELECT node2 AS node, MIN(LEAST(node1, node2)) OVER(PARTITION BY node2) AS root_id FROM recursive_union
  )
  GROUP BY node
),
-- 聚合用户总播放量
plays_agg AS (
  SELECT
    m.UNIQUE_ID,
    SUM(u.VIDEO_PLAYS) AS PLAYS
  FROM USER_PLAYS u
  LEFT JOIN node_mapping m
    ON CAST(u.USER_ID AS STRING) = m.node OR u.EMAIL = m.node
  GROUP BY m.UNIQUE_ID
),
-- 聚合用户总净销售额
sales_agg AS (
  SELECT
    m.UNIQUE_ID,
    SUM(s.NET_SALE) AS NET_SALE
  FROM Sales s
  LEFT JOIN node_mapping m
    ON s.EMAIL = m.node
  GROUP BY m.UNIQUE_ID
)
-- 关联输出最终结果
SELECT
  p.UNIQUE_ID,
  p.PLAYS,
  COALESCE(s.NET_SALE, 0) AS NET_SALE
FROM plays_agg p
LEFT JOIN sales_agg s
  ON p.UNIQUE_ID = s.UNIQUE_ID

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