如何处理BigQuery数据管道中的RuntimeException异常

问题场景 #

当使用BigQueryIO写入数据时，配置了.withCreateDisposition(CreateDisposition.CREATE_NEVER)，意味着目标表不存在时不会自动创建。此时尝试向不存在的表写入数据会触发PipelineExecutionException包裹的RuntimeException，且无法通过常规的try/catch捕获这个异常——因为Dataflow的执行模型是分布式的，异常在Runner执行阶段抛出，而非构建Pipeline的代码块。

用到的代码示例：

var write = mypipline.apply("Write table", BigQueryIO
    .<Employee>write()
    .to(targetTableName_notpresent)
    .withExtendedErrorInfo()
    .withFormatFunction(new EmployeeToTableRow())
    .withSchema(schema)
    .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())
    .withTableDescription("My Test Table")
    .withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS)
    .withCreateDisposition(CreateDisposition.CREATE_NEVER)
    .withWriteDisposition(WriteDisposition.WRITE_APPEND));

触发的异常信息：

org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.RuntimeException: com.google.api.client.googleapis.json.GoogleJsonResponseException: 404 Not Found
POST https://bigquery.googleapis.com/bigquery/v2/projects/XXXX/datasets/jupyter/tables/not_here/insertAll?prettyPrint=false
{
  "code" : 404,
  "errors" : [ {
    "domain" : "global",
    "message" : "Not found: Table XXXX:jupyter.not_here",
    "reason" : "notFound"
  } ],
  "message" : "Not found: Table XXXX:jupyter.not_here",
  "status" : "NOT_FOUND"
}
    at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:373)
    at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:341)
    at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:218)
    at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:67)
    at org.apache.beam.sdk.Pipeline.run(Pipeline.java:323)
    at org.apache.beam.sdk.Pipeline.run(Pipeline.java:309)
    at .(#126:1)

可行解决方案 #

1. 提前检查表是否存在 #

在构建Pipeline之前，通过BigQuery客户端API主动检查表的存在性，从源头避免触发运行时异常：

// 初始化BigQuery客户端
BigQuery bigQuery = BigQueryOptions.getDefaultInstance().getService();
TableId tableId = TableId.of(projectId, datasetId, tableName);

// 检查表是否存在
if (!bigQuery.getTable(tableId).exists()) {
    // 自定义处理逻辑：比如终止Pipeline、记录告警、手动创建表（如果允许）等
    System.err.println("目标表不存在，终止执行");
    return;
}

// 继续构建并运行Pipeline
var write = mypipline.apply("Write table", BigQueryIO...);

2. 使用错误输出分支捕获写入错误 #

对于STREAMING_INSERTS模式，可以通过withFailedRecordsPCollection将写入失败的记录（包括表不存在导致的批量失败）输出到单独的PCollection，实现单条记录级别的错误处理：

// 定义失败记录的输出标签
TupleTag<Employee> failedRecordsTag = new TupleTag<Employee>() {};

// 执行写入并获取主输出和错误输出
PCollectionTuple results = mypipline.apply("Write table", BigQueryIO
    .<Employee>write()
    .to(targetTableName_notpresent)
    .withExtendedErrorInfo()
    .withFormatFunction(new EmployeeToTableRow())
    .withSchema(schema)
    .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())
    .withTableDescription("My Test Table")
    .withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS)
    .withCreateDisposition(CreateDisposition.CREATE_NEVER)
    .withWriteDisposition(WriteDisposition.WRITE_APPEND)
    // 指定失败记录的输出标签
    .withFailedRecordsPCollection(failedRecordsTag));

// 处理失败的记录
results.get(failedRecordsTag).apply("Handle failed writes", ParDo.of(new DoFn<Employee, Void>() {
    @ProcessElement
    public void processElement(ProcessContext c) {
        Employee failedEmp = c.element();
        // 获取关联的错误信息（需配合withExtendedErrorInfo）
        Map<String, String> errorInfo = c.sideInput(BigQueryIO.Write.getFailedRecordsErrorInfoTag());
        String errorMsg = errorInfo.get(failedEmp.toString()); // 根据实际业务调整关联逻辑
        // 自定义处理：记录日志、发送告警、存入错误表等
        System.err.printf("写入失败，记录：%s，错误：%s%n", failedEmp, errorMsg);
    }
}).withSideInputs(BigQueryIO.Write.getFailedRecordsErrorInfoTag()));

注意：表不存在属于非临时错误，重试策略不会触发重试，所有写入请求会直接进入失败分支。

3. 捕获Pipeline执行阶段的全局异常 #

虽然无法在构建Pipeline时用try/catch捕获，但可以在调用pipeline.run().waitUntilFinish()时包裹try/catch，捕获PipelineExecutionException后判断错误类型：

try {
    PipelineResult result = mypipline.run().waitUntilFinish();
    if (result.getState() != PipelineResult.State.DONE) {
        System.err.println("Pipeline执行未完成，状态：" + result.getState());
    }
} catch (PipelineExecutionException e) {
    // 拆解异常链，定位原始错误
    Throwable cause = e.getCause();
    if (cause instanceof RuntimeException && cause.getCause() instanceof GoogleJsonResponseException) {
        GoogleJsonResponseException gEx = (GoogleJsonResponseException) cause.getCause();
        if (gEx.getStatusCode() == 404 && gEx.getDetails().getErrors().stream()
                .anyMatch(err -> "notFound".equals(err.getReason()) && err.getMessage().contains("Table"))) {
            // 处理表不存在的逻辑
            System.err.println("检测到目标表不存在：" + gEx.getMessage());
        } else {
            // 处理其他BigQuery错误
            System.err.println("BigQuery写入错误：" + gEx.getMessage());
        }
    } else {
        // 处理其他执行异常
        e.printStackTrace();
    }
}

这种方式适合做全局异常告警或终止处理，但无法针对单条记录做精细化处理。

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