Apache Beam RetryTransientErrors/neverRetry 不遵守 table 未找到错误

Question

下面是我用来将数据写入 BigQuery 的代码

WriteResult result = formattedData.get(successRows).setCoder(TableRowJsonCoder.of())
            .apply("BQ SteamingInserts",BigQueryIO.writeTableRows()
                    .withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS)
                    .withFormatFunction(new TableRowFormatFn())
                    .to(new DestinationMapper())
                    .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                    .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                    .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())
                    .withoutValidation()
                    .withExtendedErrorInfo());

代码正在处理所有与模式相关的问题，但是当 BigQuery 中不存在 table 时，它会无限期地重试插入，导致管道停滞。

下面是Dataflow中得到的错误

java.lang.RuntimeException: com.google.api.client.googleapis.json.GoogleJsonResponseException: 404 Not Found
{
  "code" : 404,
  "errors" : [ {
    "domain" : "global",
    "message" : "Not found: Table analytics-and-presentation:trusted_layer_ods.wrong_table",
    "reason" : "notFound"
  } ],
  "message" : "Not found: Table analytics-and-presentation:trusted_layer_ods.wrong_table",
  "status" : "NOT_FOUND"
}

有人可以帮忙吗？

Answer 1

这看起来像是一种预期行为，因为在流式管道的情况下，Dataflow 将无限期地重新进入。在批处理管道中尝试 4 次后，该作业将失败。

您必须在代码中明确定义您期望实现的目标。您可以从 official Google Cloud Platform github page.

上的示例中获得灵感

使用您当前的代码，您必须确保提前创建 table 以避免此错误。

Apache Beam RetryTransientErrors/neverRetry 不遵守 table 未找到错误

Apache Beam RetryTransientErrors/neverRetry does not respect table not found error

google-cloud-dataflow

apache-beam