运行DataFlow作业时BigQuery重复记录问题排查

核心原因 #

你的方案存在几个关键漏洞，导致重复记录无法被彻底过滤：

1. 双读取的时间窗口不一致 #

你分开读取源表记录和目标表哈希，这两个操作是独立的有界快照读取。比如：

• 作业在T0时刻读取目标表的哈希列表
• 作业运行期间（T0-T1），上一次作业的写入刚完成，或者其他并行作业写入了新记录
• 当前作业在T1处理源表记录时，这些新增的哈希不在之前读取的列表里，导致重复写入

这种情况下，两个数据集的读取时间点不同，过滤逻辑基于的是旧的哈希快照，自然会漏掉新写入的重复项。

2. 内存加载哈希的风险 #

用beam.pvalue.AsIter(hashes)把所有哈希加载到每个Worker内存里，存在两个问题：

• 如果目标表哈希量很大，会导致Worker内存不足，甚至哈希列表被截断，部分已存在的哈希没被过滤
• 哈希列表是作业启动时的静态快照，无法感知作业运行期间目标表的新增数据

3. 写入阶段的重试重复 #

DataFlow的容错机制会在Worker失败时重试任务。如果WriteToBigQuery没有开启幂等写入，已经成功写入的记录会被重试任务再次写入，形成重复。

4. 哈希生成的潜在问题（需验证） #

如果add_hash_field函数存在逻辑漏洞，比如：

• 字段顺序不固定，导致同一条记录生成不同哈希
• 空值、特殊字符处理不一致，生成不稳定的哈希
• 使用了非确定性的哈希算法
  也会导致过滤失效或误判，但这种情况更多是出现"漏去重"而非重复写入。

针对性解决办法 #

1. 改用SQL层面原子过滤（最有效） #

把源表读取和已存在哈希的过滤合并成一个BigQuery查询，利用BigQuery的原子快照特性，保证源表和目标表的读取是同一个时间点的状态：

read_from_source_query = """
WITH source_records AS (
    -- 替换成你的源表查询逻辑，同时生成哈希
    SELECT 
        field1, field2, field3,
        GENERATE_HASH(field1, field2, field3) AS HashValue
    FROM `your-project.source-dataset.source-table`
    -- 加上源表过滤条件，比如按时间范围取最近1小时数据
    WHERE event_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
), existing_hashes AS (
    SELECT HashValue FROM `your-project.target-dataset.target-table`
)
SELECT * FROM source_records
WHERE HashValue NOT IN (SELECT HashValue FROM existing_hashes)
"""

# 直接读取过滤后的记录，省去DataFlow内的哈希过滤步骤
records = p | 'Read Filtered Records' >> beam.io.Read(
    beam.io.ReadFromBigQuery(query=read_from_source_query, use_standard_sql=True)
)

2. 修复GroupByKey后的写入幂等性 #

给WriteToBigQuery开启幂等写入，避免重试导致的重复：

records_to_store | 'Write to target table' >> beam.io.WriteToBigQuery(
    target_table,
    write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
    create_disposition=beam.io.BigQueryDisposition.CREATE_NEVER,
    # 开启幂等写入，基于记录的哈希或唯一键去重
    idempotent_writes=True,
    # 配置重试策略，只重试临时错误
    insert_retry_strategy=beam.io.gcp.bigquery_tools.RetryStrategy.RETRY_ON_TRANSIENT_ERROR
)

3. 验证哈希生成逻辑 #

检查add_hash_field函数，确保：

• 固定参与哈希的字段顺序，比如始终按field1, field2, field3的顺序计算
• 统一处理空值，比如把NULL转成字符串"NULL"再参与哈希
• 使用稳定的哈希算法，比如MD5或SHA256，避免使用非确定性算法

4. 给目标表添加唯一约束（兜底） #

在BigQuery目标表上给HashValue字段添加唯一约束，即使DataFlow层面出现漏网之鱼，BigQuery也会拦截重复写入：

ALTER TABLE `your-project.target-dataset.target-table`
ADD CONSTRAINT unique_hash UNIQUE(HashValue)

注：BigQuery的唯一约束默认是软约束，可通过设置enforce_unique_key=true开启强制校验（需注意性能影响）。

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