使用Matillion ETL迁移GA4数据至Snowflake：如何避免重复记录

针对多次运行Matillion任务导致Snowflake表出现重复记录的问题，可通过以下几种实用方案解决：

方案1：用Matillion Merge组件替代直接插入 #

这是最直接的去重方式，通过匹配唯一记录避免重复插入：

• 在Snowflake编排任务中，移除原有的Insert组件，替换为Merge组件。
• 确定GA4事件的唯一标识键：GA4每条事件记录可通过user_pseudo_id+event_id+event_timestamp的组合唯一确定（这三个字段的组合不会重复）。
• 配置Merge逻辑：设置当源数据（从BigQuery拉取的记录）与目标表的唯一键匹配时跳过，不匹配时执行插入。
• 核心SQL逻辑示例：

MERGE INTO your_snowflake_ga4_table tgt
USING (
  -- 原BigQuery查询语句
  SELECT * FROM events_* 
  WHERE _table_suffix = FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 2 day))
) src
ON tgt.user_pseudo_id = src.user_pseudo_id
AND tgt.event_id = src.event_id
AND tgt.event_timestamp = src.event_timestamp
WHEN NOT MATCHED THEN INSERT (user_pseudo_id, event_id, event_timestamp, /* 其他所有字段 */)
VALUES (src.user_pseudo_id, src.event_id, src.event_timestamp, /* 对应源字段 */);

方案2：在BigQuery查询阶段提前去重 #

如果同一份数据在BigQuery中可能存在重复，或多次拉取时重复获取，可在查询时先过滤重复记录：

• 修改Matillion中BigQuery的查询语句，加入DISTINCT关键字：

SELECT DISTINCT * 
FROM events_* 
WHERE _table_suffix = FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 2 day));

• 若需要更严谨的去重（比如某些字段存在微小差异），可基于唯一键分组聚合：

SELECT user_pseudo_id, event_id, event_timestamp, 
       MAX(event_name) AS event_name, 
       ANY_VALUE(params) AS params
       -- 其他字段根据业务需求选择聚合方式，GA4事件字段大多适合取MAX/ANY_VALUE
FROM events_* 
WHERE _table_suffix = FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 2 day))
GROUP BY user_pseudo_id, event_id, event_timestamp;

方案3：用控制表记录已迁移的分区/记录 #

通过维护一张迁移日志表，避免重复拉取已处理过的数据：

1. 在Snowflake创建控制表ga4_migration_log，字段包括dataset_id（区分两个网站的数据集）、partition_date（BigQuery表的日期后缀）、migrated_at（迁移时间）。
2. 修改Matillion任务流程：
   • 先查询控制表，获取已迁移的dataset_id和partition_date组合。
   • 在BigQuery查询中排除这些已处理的记录（假设两个网站的数据在不同BigQuery数据集下）：

     SELECT * 
     FROM `your_bq_project.dataset1.events_*` 
     WHERE _table_suffix = FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 2 day))
     AND ('dataset1', _table_suffix) NOT IN (SELECT dataset_id, partition_date FROM ga4_migration_log)
     UNION ALL
     SELECT * 
     FROM `your_bq_project.dataset2.events_*` 
     WHERE _table_suffix = FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 2 day))
     AND ('dataset2', _table_suffix) NOT IN (SELECT dataset_id, partition_date FROM ga4_migration_log);
     

   • 迁移任务执行成功后，将本次处理的dataset_id和partition_date插入控制表。

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