数据去重逻辑设计

以下为按 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 之后）。
- 无稳定键且多列完全并列时，结果不具有跨运行的确定性，需在源端或摄取层补充唯一键。

以下为按 sku + title 去重，并保留“最近30天成交数 sales_30d 最大；并列时上架时间 last_on_shelf 最新”的数据去重逻辑与可直接使用的 SQL 模板。逻辑完全基于窗口函数排名，确保结果稳定、可复现。

一、排序与选择规则
- 分组键：sku, title
- 排序优先级：
  1) sales_30d 降序（非空优先）
  2) last_on_shelf 降序（非空优先）
  3) 可选的最终稳定排序键（如 id 或 updated_at），用于打破所有字段完全相同的并列

二、通用 ANSI SQL（适用于大多数数据库）
说明：使用 ROW_NUMBER() + 子查询；使用 CASE WHEN 将 NULL 放到最后，避免 NULLS LAST 不同方言兼容性问题。

WITH ranked AS (
  SELECT
    t.*,
    ROW_NUMBER() OVER (
      PARTITION BY t.sku, t.title
      ORDER BY
        CASE WHEN t.sales_30d IS NULL THEN 1 ELSE 0 END ASC,  -- 非空优先
        t.sales_30d DESC,
        CASE WHEN t.last_on_shelf IS NULL THEN 1 ELSE 0 END ASC,
        t.last_on_shelf DESC
        -- 可选：增加稳定排序键，避免完全并列的不确定性
        -- , t.id ASC
    ) AS rn
  FROM product_items t
)
SELECT *
FROM ranked
WHERE rn = 1;

三、PostgreSQL（使用 DISTINCT ON，性能与可读性更优）
SELECT DISTINCT ON (t.sku, t.title)
  t.*
FROM product_items t
ORDER BY
  t.sku, t.title,
  t.sales_30d DESC NULLS LAST,
  t.last_on_shelf DESC NULLS LAST
  -- 可选稳定排序键
  -- , t.id ASC
;

四、BigQuery / Snowflake（支持 QUALIFY，语句更简洁）
BigQuery:
SELECT
  t.*
FROM product_items t
QUALIFY ROW_NUMBER() OVER (
  PARTITION BY t.sku, t.title
  ORDER BY
    IF(t.sales_30d IS NULL, 1, 0), t.sales_30d DESC,
    IF(t.last_on_shelf IS NULL, 1, 0), t.last_on_shelf DESC
    -- 可选稳定排序键
    -- , t.id
) = 1;

Snowflake:
SELECT
  t.*
FROM product_items t
QUALIFY ROW_NUMBER() OVER (
  PARTITION BY t.sku, t.title
  ORDER BY
    IFF(t.sales_30d IS NULL, 1, 0), t.sales_30d DESC,
    IFF(t.last_on_shelf IS NULL, 1, 0), t.last_on_shelf DESC
    -- 可选稳定排序键
    -- , t.id
) = 1;

五、Spark SQL（Databricks / Hive 方言）
WITH ranked AS (
  SELECT
    t.*,
    ROW_NUMBER() OVER (
      PARTITION BY t.sku, t.title
      ORDER BY
        CASE WHEN t.sales_30d IS NULL THEN 1 ELSE 0 END,
        t.sales_30d DESC,
        CASE WHEN t.last_on_shelf IS NULL THEN 1 ELSE 0 END,
        t.last_on_shelf DESC
        -- 可选稳定排序键
        -- , t.id
    ) AS rn
  FROM product_items t
)
SELECT *
FROM ranked
WHERE rn = 1;

六、物化与性能建议
- 写回去重结果：
  - 覆盖写：CREATE TABLE product_items_dedup AS <上述查询>;
  - 或 MERGE INTO 目标表（以 sku, title 为匹配键）执行幂等更新。
- 索引/分桶/聚簇：
  - OLTP（如 PostgreSQL/MySQL）：建立复合索引 (sku, title)。
  - 数仓（如 BigQuery/Snowflake）：CLUSTER BY/CLUSTERING KEY (sku, title) 以优化分组与扫描。

七、注意事项
- 确认 sales_30d、last_on_shelf 的类型分别为数值与时间戳；若为字符串，请先做显式转换。
- 若存在完全并列（sales_30d 与 last_on_shelf 都相同），建议在 ORDER BY 末尾加入稳定排序键（如 id 或 updated_at），确保结果确定性。
- 若业务上需特殊处理 NULL（例如将 NULL 视为 0 或视为最旧时间），需相应调整 CASE WHEN 或使用 COALESCE。