google-cloud-dataflow - 使用 Dataflow failed insert WriteResult 处理 table not found 异常

Question

您好，我想在 Dataflow pipelnie 中动态动态创建表

首先，我捕获 BigQueryIO WriteResult，然后使用它来创建表

        WriteResult writeResult =
            incomingRecords.apply(
                    "WriteToBigQuery",
                    BigQueryIO.<TableRowWithSchema>write()
                            .to(new DynamicTables())
                            .withFormatFunction(TableRowWithSchema::getTableRow)
                            .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                            .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                            .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors()));

但是还是遇到table not found异常

Caused by: org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.RuntimeException:     com.google.api.client.googleapis.json.GoogleJsonResponseException: 404 Not Found
{
  "code" : 404,
  "errors" : [ {
  "domain" : "global",
​  "message" : "Not found: Table projectId:datasetId.StreamComment",
  "reason" : "notFound"
} ],
  "message" : "Not found: Table projectId:datasetId.StreamComment",
  "status" : "NOT_FOUND"
}

有什么问题吗？谢谢

Answer 1

你有两处错误:

1) 如果您想动态创建表，您需要使用CREATED_IF_NEEDED 处置。

withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)

2)动态创建表时，需要传递Schema，使用withSchema方法或withJsonSchema方法。

这样就可以了!