数据去重逻辑设计

以下为按 user_id + email 去重的推荐实现与注意事项。规则为：同一(user_id, email)分组内，优先保留 event_time 最新；若 event_time 并列，优先 login_count 高；如仍并列，使用稳定主键或摄取时间作为最终决策键。

一、字段与预处理假设

• 必需字段：user_id, email, event_time(TIMESTAMP), login_count(INT)
• 建议存在的稳定决策键：id(自增或唯一事件ID)或 ingest_time
• 规范化建议：
  • 对 email 做标准化，示例：LOWER(TRIM(email))，避免大小写或空白差异造成重复未对齐
  • NULL 处理：
    • 将 event_time 为 NULL 视为最旧（排在最后）
    • 将 login_count 为 NULL 视为 0 或按业务定义；下例用 NULLS LAST 逻辑处理

二、通用 SQL 去重方案（窗口函数，推荐） 说明：使用窗口函数按(user_id, email_norm)分区，根据优先级排序并取 row_number=1。

标准写法（兼容性较好，无需 QUALIFY）： WITH normalized AS ( SELECT user_id, LOWER(TRIM(email)) AS email_norm, event_time, login_count, id FROM user_events ), ranked AS ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY user_id, email_norm ORDER BY CASE WHEN event_time IS NULL THEN 1 ELSE 0 END, -- NULL 放最后 event_time DESC, CASE WHEN login_count IS NULL THEN 1 ELSE 0 END, -- NULL 放最后 login_count DESC, id DESC -- 最终稳定决策键（如无可去掉，但将非确定） ) AS rn FROM normalized ) SELECT user_id, email_norm AS email, event_time, login_count, id FROM ranked WHERE rn = 1;

Snowflake/BigQuery 等支持 QUALIFY 的写法： SELECT user_id, LOWER(TRIM(email)) AS email, event_time, login_count, id FROM user_events QUALIFY ROW_NUMBER() OVER ( PARTITION BY user_id, LOWER(TRIM(email)) ORDER BY event_time DESC NULLS LAST, login_count DESC NULLS LAST, id DESC ) = 1;

三、写回目标表的 MERGE（示例）

• 场景：将去重结果落入 user_events_dedup。这里使用一个临时 CTE ranked 先筛出 rn=1，再 MERGE。 WITH normalized AS ( SELECT user_id, LOWER(TRIM(email)) AS email_norm, event_time, login_count, id FROM user_events ), ranked AS ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY user_id, email_norm ORDER BY CASE WHEN event_time IS NULL THEN 1 ELSE 0 END, event_time DESC, CASE WHEN login_count IS NULL THEN 1 ELSE 0 END, login_count DESC, id DESC ) AS rn FROM normalized ) MERGE INTO user_events_dedup t USING (SELECT * FROM ranked WHERE rn = 1) s ON (t.user_id = s.user_id AND t.email = s.email_norm) WHEN MATCHED THEN UPDATE SET event_time = s.event_time, login_count = s.login_count, id = s.id WHEN NOT MATCHED THEN INSERT (user_id, email, event_time, login_count, id) VALUES (s.user_id, s.email_norm, s.event_time, s.login_count, s.id);

四、Spark（DataFrame）实现 from pyspark.sql import functions as F, Window as W

dfn = (df .withColumn("email_norm", F.lower(F.trim(F.col("email")))))

w = (W.partitionBy("user_id", "email_norm") .orderBy(F.col("event_time").desc_nulls_last(), F.col("login_count").desc_nulls_last(), F.col("id").desc()))

df_dedup = (dfn .withColumn("rn", F.row_number().over(w)) .filter(F.col("rn") == 1) .drop("rn"))

五、性能与稳定性建议

• 稳定决策键：如存在完全相同 event_time 与 login_count 的记录，务必使用稳定且单调的 id 或 ingest_time 做最终排序，保证结果确定性。
• 索引/分区：
  • 事务型数据库：为(user_id, email_norm)建组合索引；如支持可在索引中加入 event_time DESC, login_count DESC 提升排序性能。
  • 数据仓库：对(user_id, email_norm)进行分桶/聚簇（如 BigQuery clustering, Snowflake clustering key）。
• 预聚合优化（可选，大数据量时）：先按(user_id, email_norm)求 max(event_time)，筛出候选，再在候选中取 max(login_count)，最后用稳定键打破并列，减少窗口数据量。

六、边界与数据质量

• email 规范化策略需与业务确认（大小写、空白、别名等）。
• 若 event_time 存在时区，统一到 UTC。
• 明确 login_count 的 NULL 业务含义（若表示未知，可与 0 区分；上面示例将其排在非 NULL 之后）。
• 无稳定键且多列完全并列时，结果不具有跨运行的确定性，需在源端或摄取层补充唯一键。