Apache Beam RetryTransientErrors / neverRetry не учитывает ошибку "таблица не найдена"

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, он продолжает повторять попытки вставки бесконечнов результате чего трубопровод останавливается.

Ниже приведена ошибка, полученная в потоке данных

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

Это выглядит как ожидаемое поведение, так как поток данных будет бесконечно возвращаться в случае потокового конвейера.Задание было бы неудачным после 4 попыток в пакетном конвейере.

Вы должны четко определить в своем коде, чего вы ожидаете достичь.Вы можете черпать вдохновение из примеров, найденных на официальной странице github Google Cloud Platform .

. При наличии настоящего кода вы должны заранее создать таблицу, чтобы избежать этой ошибки.