如何确保Dataflow侧输入刷新时不丢失数据？

问题根源分析 #

1. 窗口触发器配置不合理：当前使用AfterProcessingTime.pastFirstElementInPane()触发器，会在GenerateSequence的触发元素到达后立即触发窗口计算，此时BigQuery查询可能还未完成，导致窗口中无数据或仅部分数据，生成不完整的侧输入Map。discardingFiredPanes会丢弃已触发窗格的数据，后续查询输出的完整数据会进入新窗格，但侧输入可能未及时更新到最新完整数据。
2. BigQuery查询无完整性校验：现有代码未对查询结果的完整性做验证，若遇到网络波动、BigQuery临时错误，可能导致部分结果未被捕获且无重试逻辑，直接丢失数据。
3. 侧输入更新时机不匹配：触发逻辑未等待BigQuery查询完全完成就生成Map，导致缓存的数据始终是不完整的中间状态。

具体修复方案 #

方案1：调整窗口触发器，等待查询完成后生成侧输入 #

修改窗口触发逻辑，确保只有当所有查询结果输出到窗口后，才触发Map生成。可以选择两种方式：

方式A：使用默认触发器+允许延迟

PCollection<TableRow> sideInput = pipeline
    .apply(GenerateSequence.from(0).withRate(1, Duration.standardSeconds(180)))
    .apply(Window.<Long>into(new GlobalWindows())
        // 用默认的AfterWatermark触发器，确保所有数据到达窗口后再计算
        .withAllowedLateness(Duration.standardMinutes(5)) // 预留足够处理延迟
        .discardingFiredPanes())
    .apply("REFRESH-SIDEINPUT", ParDo.of(new SideInputQuery(psToBqOptions.getBQProjectSideInput()))); 

PCollectionView<Map<String, String>> sideInputView = sideInput
    .apply("To map", ParDo.of(new TableRowToMap()))
    .apply(View.<String, String>asMap());

方式B：添加处理时间延迟（根据实际查询耗时调整）

.apply(Window.<Long>into(new GlobalWindows())
    .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane()
        .plusDelayOf(Duration.standardMinutes(2)))) // 设置足够覆盖查询耗时的延迟
    .discardingFiredPanes())

方案2：强化BigQuery查询的完整性保障 #

在SideInputQuery中添加重试逻辑和结果完整性校验，避免数据丢失：

@ProcessElement
public void processElement(ProcessContext c) throws Exception {
    BigQuery bigQuery = BigQueryOptions.getDefaultInstance().getService();
    QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder("SELECT distinct storeId as storeID_rebuild, "
                    + "cast(C.code as INT64) as storeId  "
                    + "FROM `" + ProjectIdRefresh.get() + ".{dataser}.{table}` s INNER JOIN UNNEST(s.{field}) as C on (C.type='type') "
                    + "where 1=1 ").build();

    // 重试机制处理临时错误
    TableResult results = null;
    int retryCount = 3;
    while (retryCount > 0) {
        try {
            results = bigQuery.query(queryConfig);
            if (results.getTotalRows() == null || results.getTotalRows() == 0) {
                LOG.warn("查询无结果，重试中...");
                retryCount--;
                Thread.sleep(1000);
                continue;
            }
            break;
        } catch (BigQueryException e) {
            LOG.error("BigQuery查询失败，重试中...", e);
            retryCount--;
            Thread.sleep(2000);
            if (retryCount == 0) throw e;
        }
    }

    if (results == null) {
        LOG.error("多次重试后仍未获取有效查询结果");
        return;
    }

    long totalRows = results.getTotalRows();
    long processedRows = 0;
    for(FieldValueList fvl : results.iterateAll()) {
        processedRows++;
        TableRow tr = new TableRow();
        tr.set("storeId", fvl.get("storeId").getNumericValue());
        tr.set("storeID_rebuild", fvl.get("storeID_rebuild").getNumericValue());
        LOG.info("storeId: {}, storeID_rebuild: {}", 
            fvl.get("storeId").getNumericValue(), 
            fvl.get("storeID_rebuild").getNumericValue());
        c.output(tr);           
    }

    // 校验处理行数与查询总行数是否一致
    if (processedRows != totalRows) {
        LOG.error("处理行数{}与查询返回总行数{}不匹配，数据可能不完整！", processedRows, totalRows);
        throw new RuntimeException("查询结果处理不完整");
    }
}

方案3：大场景优化——文件中转法（适用于超大量查询结果） #

如果查询结果数据量极大，直接在ParDo中处理易引发内存问题，可以改为：

1. 每次触发时，将BigQuery查询结果原子性导出到GCS存储桶
2. 使用Dataflow的TextIO/AvroIO读取GCS文件，转换为Map后生成侧输入
   这种方式依赖GCS文件的原子性，确保只有完整的导出文件才会被读取，彻底避免数据不完整问题。

验证建议 #

• 部署修改后的管道后，观察日志中processedRows与totalRows是否一致，确认所有查询结果都被输出
• 抽样检查侧输入Map的内容，验证是否包含所有预期的映射关系
• 模拟不同数据量的刷新场景，验证数据完整性

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