Dataflow上Apache Beam作业处理Avro文件时出现重复写入记录问题

• Dataflow的「精确一次」语义仅针对有状态计算和幂等Sink的场景生效，如果管道存在非幂等自定义逻辑、或算子融合导致的重试副作用，依然会出现数据重复。
• 本次重复的核心诱因是算子融合+Bundle重试：默认Dataflow会将无shuffle操作的相邻算子融合到同一执行线程运行，你的BatchElements→RedactPIIData→FlatMap→WriteToAvro全链路无shuffle，会被合并为单个执行阶段。当该阶段因节点扩缩容、瞬时错误触发Bundle重试时，部分已经完成处理写入的元素会被重新执行全链路，最终生成重复写入。
• 自研UUID去重方案失效的原因：你是在管道运行过程中为每条记录生成随机UUID，重试时同一条原始记录会生成全新的UUID，GroupByKey无法识别为同一条数据，因此无法实现去重，本地运行无重试所以可以生效。
• 实验性Reshuffle算子生效的原因：Reshuffle会触发shuffle操作，强制打断算子融合，将链路拆分为两个独立执行阶段，前一阶段的处理结果会持久化到shuffle存储，重试时仅会从shuffle存储读取数据处理后半段，不会重新读取原始Avro重跑全链路，因此避免重复。

方案1：用内容哈希shuffle替代实验性Reshuffle（推荐） #

自行实现稳定的shuffle操作打断融合，同时基于原始内容哈希实现天然去重，无额外依赖且完全生产可用，示例代码如下：

def generate_content_hash(record):
    # 基于原始记录全量字段生成唯一哈希，重试时同一条记录的哈希不会变化
    import hashlib
    sorted_record_str = str(sorted(record.items())).encode("utf-8")
    return (hashlib.sha256(sorted_record_str).hexdigest(), record)

read_from_avro \
    | 'Group into batches' >> beam.BatchElements(700) \
    | "RedactData" >> RedactPIIData(project, redact_fields) \
    | beam.FlatMap(lambda elements: elements)
    # 新增如下3行代码即可解决问题
    | '生成内容哈希' >> beam.Map(generate_content_hash)
    | '按哈希分组' >> beam.GroupByKey()
    | '按哈希去重' >> beam.FlatMap(lambda key_val: [key_val[1][0]])
    | '写入Avro' >> WriteToAvro(...)

方案2：优化自定义DoFn的幂等性 #

为RedactPIIData自定义DoFn增加幂等校验逻辑：为每个输入batch生成基于内容的唯一标识，处理成功后缓存该标识对应的处理结果，重试时如果命中缓存直接返回结果，避免重复处理输出重复数据。

方案3：禁用算子融合（临时验证用，不推荐生产使用） #

如果修改代码成本较高，可以在提交Dataflow作业时添加参数--experiments=no_use_fused_runner，强制关闭算子融合，每个算子独立执行，避免融合导致的重试重复，但该参数会显著降低作业执行性能，仅适合临时验证问题使用。

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