Snowflake SQL查询：基于邮箱/手机号识别唯一用户并生成Person ID

这是典型的连通组件识别问题：通过邮箱或手机号的关联关系，将属于同一自然人的所有customer_id归为一组，最终用组内最小的customer_id作为该组的唯一person_id。以下提供两种可行的Snowflake SQL实现方案：

方案一：递归CTE（逻辑直观，适合中小数据量） #

通过递归遍历所有关联的客户，逐步合并连通组并记录组内最小ID：

WITH RECURSIVE customer_links AS (
    -- 锚点：每个客户初始自成一组，最小ID为自身
    SELECT 
        customer_id AS current_id,
        customer_id AS min_person_id,
        customer_email,
        customer_phone_number
    FROM customer

    UNION ALL

    -- 递归：通过邮箱/手机号匹配关联客户，合并连通组并更新最小ID
    SELECT 
        c.customer_id AS current_id,
        LEAST(cl.min_person_id, c.customer_id) AS min_person_id,
        c.customer_email,
        c.customer_phone_number
    FROM customer c
    JOIN customer_links cl 
        ON (c.customer_email = cl.customer_email OR c.customer_phone_number = cl.customer_phone_number)
        AND c.customer_id <> cl.current_id
        -- 限制ID大小避免循环和重复计算
        AND c.customer_id > cl.current_id
),
final_person_ids AS (
    -- 提取每个客户对应的最小组ID
    SELECT 
        current_id AS customer_id,
        MIN(min_person_id) AS person_id
    FROM customer_links
    GROUP BY current_id
)
-- 关联原表输出完整结果
SELECT 
    c.customer_id,
    c.customer_email,
    c.customer_phone_number,
    f.person_id
FROM customer c
JOIN final_person_ids f ON c.customer_id = f.customer_id
ORDER BY c.customer_id;

逻辑说明： #

1. 锚点阶段：将每个客户单独作为初始组，记录自身ID为组内最小ID。
2. 递归阶段：通过邮箱或手机号匹配关联客户，仅处理未循环的关联对（c.customer_id > cl.current_id），每次合并时取两组的最小ID作为新的组最小ID。
3. 最终聚合：对每个客户，取其所有关联记录中的最小ID，即为该客户所属的person_id。

方案二：Snowflake图函数（高效，适合大数据量） #

利用Snowflake内置的图处理功能GRAPH_PATH，快速识别连通组件并计算组内最小ID：

WITH customer_graph AS (
    -- 构建无向边：共享邮箱/手机号的客户对建立连接（避免双向重复边）
    SELECT 
        c1.customer_id AS src,
        c2.customer_id AS dst
    FROM customer c1
    JOIN customer c2 
        ON (c1.customer_email = c2.customer_email OR c1.customer_phone_number = c2.customer_phone_number)
        AND c1.customer_id < c2.customer_id
),
connected_components AS (
    -- 识别连通组件并计算组内最小ID
    SELECT 
        src AS customer_id,
        MIN(MIN_VALUE) OVER (PARTITION BY component_id) AS person_id
    FROM TABLE(
        GRAPH_PATH(
            -- 顶点表：所有客户ID
            (SELECT DISTINCT customer_id AS vertex FROM customer),
            -- 边表：预定义的客户关联边
            customer_graph,
            -- 遍历起点：所有客户
            (SELECT DISTINCT customer_id AS start FROM customer),
            -- 最大路径长度：设置足够大以覆盖所有连通关系
            100,
            -- 无向图模式：关联关系是双向的
            TRUE
        )
    )
)
-- 输出最终结果
SELECT 
    c.customer_id,
    c.customer_email,
    c.customer_phone_number,
    cc.person_id
FROM customer c
JOIN connected_components cc ON c.customer_id = cc.customer_id
ORDER BY c.customer_id;

逻辑说明： #

1. 构建图边：为每对共享邮箱/手机号的客户建立单向边（c1.customer_id < c2.customer_id），避免重复计算。
2. 图遍历：GRAPH_PATH函数遍历所有连通组件，返回每个节点所属的组件ID及路径中的节点值。
3. 计算组最小ID：通过窗口函数对每个组件的节点值取最小值，得到该组件的person_id。

两种方案均能输出与示例一致的结果：customer_id 1-5的person_id为1，customer_id 6-7的person_id为6。

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