使用 Dataflow failed insert WriteResult 处理 table not found 异常答案

【问题标题】：Use Dataflow failed insert WriteResult to handle table not found exception使用 Dataflow failed insert WriteResult 处理 table not found 异常
【发布时间】：2019-04-17 00:08:22
【问题描述】：

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

首先，我捕获 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 Exception

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"
}

有什么问题吗？谢谢

【问题讨论】：

标签： google-cloud-dataflow apache-beam dataflow apache-beam-io

【解决方案1】：

你有两件事错了：

1) 如果要动态创建表，则需要使用CREATED_IF_NEEDED 配置。

withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)

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

这样就行了！

【讨论】：