使用Dataflow写入Bigtable时，如何处理插入失败及计数差异问题？

我正在构建一个将GCS中的JSON格式数据迁移至Bigtable的Dataflow管道。管道运行正常，但Dataflow作业的写入记录数与通过BigQuery外部表查询Bigtable得到的计数不匹配。我已设置"onlyReadLatest": false以确保读取所有记录。

编辑补充：问题已定位，是生成行键的JSON值存在重复，给行键添加UUID后计数已匹配。

Dataflow写入Bigtable的核心代码：

CloudBigtableTableConfiguration bigtableTableConfig =
        new CloudBigtableTableConfiguration.Builder()
        .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
        .withTableId(options.getBigtableTableId())
            .build();       
PDone tableRows = btRow.get(successTag)
        .apply("WriteToBT", CloudBigtableIO.writeToTable(bigtableTableConfig));

BigQuery统计查询语句：

SELECT rowkey,   
    ARRAY_TO_STRING(ARRAY(SELECT value FROM UNNEST(common.id.cell)), "") AS Id,
    ARRAY_TO_STRING(ARRAY(SELECT value FROM UNNEST(common.timestamp_col.cell)), "") AS timestamp_col
FROM `<table_id>`

已知BigQuery可以通过.withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())处理插入失败，请问Bigtable是否有类似机制？若没有，该如何处理插入失败？

Bigtable的内置重试机制 #

Cloud Bigtable的Dataflow连接器（CloudBigtableIO）本身默认开启了针对瞬时错误的重试逻辑，比如网络波动、Bigtable临时限流等场景。底层依赖的Bigtable客户端会自动处理这类重试，无需额外配置。

自定义重试与失败处理 #

如果需要精细化控制或处理非瞬时错误，可以通过以下方式实现：

1. 配置客户端重试参数：
   构建CloudBigtableTableConfiguration时，可通过withConfiguration传入客户端重试配置，调整重试次数、超时时间等：

   CloudBigtableTableConfiguration bigtableTableConfig =
           new CloudBigtableTableConfiguration.Builder()
                   .withProjectId(options.getBigtableProjectId())
                   .withInstanceId(options.getBigtableInstanceId())
                   .withTableId(options.getBigtableTableId())
                   // 设置最大重试次数
                   .withConfiguration("bigtable.rpc.max.retries", "5")
                   // 设置重试超时时间（单位：毫秒）
                   .withConfiguration("bigtable.rpc.retry.timeout.ms", "30000")
                   .build();
   

2. 捕获并处理写入失败记录：
   CloudBigtableIO.writeToTable支持通过withFailureTag捕获写入失败的行，后续可对这些数据进行二次处理（比如写入死信队列、记录日志后重试）：

   TupleTag<Row> successTag = new TupleTag<Row>(){};
   TupleTag<Row> failureTag = new TupleTag<Row>(){};
   
   PCollectionTuple results = btRow.get(successTag)
           .apply("WriteToBT", CloudBigtableIO.writeToTable(bigtableTableConfig)
                   .withFailureTag(failureTag));
   
   // 将失败记录写入GCS死信桶
   results.get(failureTag)
           .apply("WriteFailedRowsToGCS", TextIO.write()
                   .to("gs://your-deadletter-bucket/failed-rows/")
                   .withSuffix(".json"));
   

3. 规避行键重复导致的数据覆盖：
   像你之前遇到的行键重复问题不属于插入失败，而是Bigtable行键唯一性特性导致后写入数据覆盖前数据。这种情况需要在预处理阶段确保行键唯一性（比如添加UUID、时间戳后缀），从根源避免数据丢失。

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