从AWS PostgreSQL导入BigQuery后表中重复数据的排查建议

一、排查重复原因 #

• 确认重复行特征：先在BigQuery中执行SELECT *, COUNT(*) AS duplicate_count FROM your_project.your_dataset.your_table GROUP BY ALL HAVING duplicate_count > 1，明确是全字段重复还是仅主键/唯一键重复（其他字段有差异）。
• 核对导入方式：确认是用BigQuery联邦查询直接同步、Cloud Dataflow增量同步、CSV/Parquet文件中转导入，还是第三方工具。不同方式的重复诱因不同：
  • 联邦查询：可能是重复执行INSERT而非MERGE；
  • 中转文件：可能是多次导出生成重复文件，或导入时未覆盖；
  • 增量同步：可能是CDC逻辑错误，将更新行识别为新行。
• 检查导入任务记录：在BigQuery控制台的“导入历史”中查看是否有重复触发的任务（比如调度重复执行、手动多次触发），APPEND模式下重复任务会直接追加数据导致重复。
• 验证源表一致性：在PostgreSQL中执行SELECT your_unique_key, COUNT(*) FROM pg_source_table GROUP BY your_unique_key HAVING COUNT(*) > 1，确认源表本身无隐藏重复（部分工具查询可能自动去重）。
• 排查数据类型转换：检查PostgreSQL与BigQuery的字段类型映射（比如timestamp时区、numeric精度），是否因类型转换导致同一行被识别为不同行。

二、消除重复与修复流程 #

1. 清理现有重复数据 #

• 全字段重复场景：直接去重覆盖原表

  CREATE OR REPLACE TABLE `your_project.your_dataset.your_table`
  AS SELECT DISTINCT * FROM `your_project.your_dataset.your_table`
  

• 唯一键重复场景：保留最新数据（假设load_time是数据导入时间）

  CREATE OR REPLACE TABLE `your_project.your_dataset.your_table`
  AS SELECT * FROM (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY your_unique_key ORDER BY load_time DESC) AS row_num
    FROM `your_project.your_dataset.your_table`
  ) WHERE row_num = 1
  

2. 修复导入流程 #

• 避免重复任务执行：如果是调度任务，添加幂等校验（比如任务执行前检查最近一次导入时间，若未超过阈值则跳过）；手动导入时确认仅触发一次。
• 替换APPEND为MERGE/覆盖模式：
  • 全量同步：将导入模式改为WRITE_TRUNCATE，每次导入前清空目标表；
  • 增量同步：使用MERGE语句基于唯一键同步，避免重复插入：

    MERGE INTO `bq_target_table` target
    USING `pg_federated_table` source
    ON target.your_unique_key = source.your_unique_key
    WHEN NOT MATCHED THEN INSERT ROW
    WHEN MATCHED THEN UPDATE SET
      col1 = source.col1,
      col2 = source.col2,
      load_time = CURRENT_TIMESTAMP()
    

• 中转文件优化：若用GCS中转，导出时添加唯一标识（如时间戳）到文件名，导入时仅读取最新文件；或导入时设置WRITE_TRUNCATE覆盖目标表。

3. 添加校验机制 #

每次导入后自动执行校验脚本，对比PostgreSQL与BigQuery的：

• 总行数：SELECT COUNT(*) FROM pg_table vs SELECT COUNT(*) FROM bq_table
• 唯一键数量：SELECT COUNT(DISTINCT your_unique_key) FROM pg_table vs SELECT COUNT(DISTINCT your_unique_key) FROM bq_table
  发现差异时及时告警，避免重复数据累积。

内容的提问来源于stack exchange，提问作者Aquiles Páez