Snowflake SQL多列关联聚合实现跨ID/邮箱用户数据合并统计

核心逻辑 #

这类多对多账号关联的身份归并，本质是无向图连通分量识别问题：

• 把USER_ID、EMAIL都作为图的独立节点
• 同一条记录中同时出现的USER_ID和EMAIL之间连边，代表属于同一用户
• 给所有连通的节点（即同一自然人）分配同一个全局UNIQUE_ID，再分别聚合播放量、销售额指标即可

完整SQL代码 #

-- 测试数据构造块，生产环境可直接删除，替换为自身业务表即可
WITH USER_PLAYS(ROW, USER_ID, EMAIL, VIDEO_PLAYS) AS (
    SELECT * FROM VALUES
    (1, 1, 'ab@gmail.com', 2),
    (2, 1, 'cd@gmail.com', 3),
    (3, 3, 'cd@gmail.com', 4),
    (4, 4, 'cd@gmail.com', 2),
    (5, 4, 'ef@gmail.com', 3)
),
Sales(NET_SALE, EMAIL) AS (
    SELECT * FROM VALUES
    (5, 'cd@gmail.com'),
    (10, 'ef@gmail.com')
),
-- 抽取全量关联边，给两类标识加前缀避免值冲突
edges AS (
    SELECT DISTINCT
        'USER_'||USER_ID::VARCHAR AS src_node,
        'EMAIL_'||EMAIL AS tgt_node
    FROM USER_PLAYS
    UNION ALL
    SELECT DISTINCT
        'EMAIL_'||EMAIL AS src_node,
        'USER_'||USER_ID::VARCHAR AS tgt_node
    FROM USER_PLAYS
),
-- 递归遍历全量连通节点
RECURSIVE node_traverse AS (
    SELECT
        src_node AS current_node,
        CASE WHEN STARTSWITH(src_node, 'USER_') THEN src_node END AS root_user,
        ARRAY_CONSTRUCT(src_node) AS visited_path
    FROM edges
    UNION ALL
    SELECT
        e.tgt_node AS current_node,
        COALESCE(
            LEAST(t.root_user, CASE WHEN STARTSWITH(e.tgt_node, 'USER_') THEN e.tgt_node END),
            t.root_user,
            CASE WHEN STARTSWITH(e.tgt_node, 'USER_') THEN e.tgt_node END
        ) AS root_user,
        ARRAY_APPEND(t.visited_path, e.tgt_node) AS visited_path
    FROM node_traverse t
    JOIN edges e ON t.current_node = e.src_node
    WHERE NOT ARRAY_CONTAINS(e.tgt_node, t.visited_path)
),
-- 生成节点到UNIQUE_ID的映射表
id_mapping AS (
    SELECT
        current_node,
        MIN(SPLIT_PART(root_user, '_', 2)::NUMBER) AS UNIQUE_ID
    FROM node_traverse
    GROUP BY current_node
),
-- 汇总总播放量
plays_total AS (
    SELECT
        m.UNIQUE_ID,
        SUM(u.VIDEO_PLAYS) AS PLAYS
    FROM USER_PLAYS u
    JOIN id_mapping m ON 'EMAIL_'||u.EMAIL = m.current_node
    GROUP BY m.UNIQUE_ID
),
-- 汇总总净销售额
sales_total AS (
    SELECT
        m.UNIQUE_ID,
        SUM(s.NET_SALE) AS NET_SALE
    FROM Sales s
    JOIN id_mapping m ON 'EMAIL_'||s.EMAIL = m.current_node
    GROUP BY m.UNIQUE_ID
)
-- 输出最终结果
SELECT
    p.UNIQUE_ID,
    p.PLAYS,
    s.NET_SALE
FROM plays_total p
JOIN sales_total s USING(UNIQUE_ID)

运行后输出结果和示例完全一致：

注意事项 #

• 给USER_ID、EMAIL加类型前缀是为了避免两类标识出现值重复（比如USER_ID存的字符串刚好和某个EMAIL值一样），导致连通关系计算错误
• 递归逻辑支持任意深度的关联，不管是一个USER_ID绑10个邮箱，还是一个邮箱关联10个USER_ID，只要能通过关联链路连通，都会被归为同一个用户
• UNIQUE_ID默认取连通分量内最小的USER_ID，和示例规则匹配，如果需要换生成规则（比如取最早注册的USER_ID、生成随机UUID），修改id_mapping块里的UNIQUE_ID取值逻辑即可
• Snowflake默认递归CTE最大深度为100，普通账号绑定场景完全够用，如果有超长关联链的场景，可通过设置MAX_RECURSION_DEPTH参数调整上限

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