在数据流中使用BigQuery beam api向Streaming_insert插入记录时请求大小错误

Question

我创建了一个google流作业，从PubSub读取并插入到BigQuery中。我使用JSON将STREAMING_INSERT数据插入到BiqQuery表中。我正面临插入问题，说明请求的大小超过了10 Mb的允许限制。数据流错误如下所示。根据我的理解，每个记录的大小是1-2Mb，根据我的理解，数据流作业将流数据插入到微批处理中，这导致了这一错误。

你能不能给我一些解决办法。

来自worker的错误消息: java.lang.RuntimeException:我们观察到一行大小为24625273字节。BigQuery支持高达10 is的请求大小，这一行太大了。您可以更改重试策略，以取消阻塞该管道，行将作为失败的插入输出。org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl$DatasetServiceImpl.insertAll(BigQueryServicesImpl.java:1088) org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl$DatasetServiceImpl.insertAll(BigQueryServicesImpl.java:1242) org.apache.beam.sdk.io.gcp.bigquery.BatchedStreamingWrite.flushRows(BatchedStreamingWrite.java:403) org.apache.beam.sdk.io.gcp.bigquery.BatchedStreamingWrite.access$900(BatchedStreamingWrite.java:67)

用于BigQuery插入的代码片段如下

.apply(
        "WriteSuccessfulRecords",
        BigQueryIO.writeTableRows().withAutoSharding()
                .withoutValidation()
                .withCreateDisposition(CreateDisposition.CREATE_NEVER)
                .withWriteDisposition(WriteDisposition.WRITE_APPEND)
                .withExtendedErrorInfo()
                .withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS)
                .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())
                .to(options.getOutputTableSpec()));

Answer 1

您的BQ写入操作正在运行到众所周知的BQ流插入API的限制中。基本上，每一批使用流插入从数据流写到BQ的数据必须小于10 in的大小。

数据流试图将批处理保持在此限制之下。但是，如果单个行大于10 may，则Dataflow不能停留在此限制范围内，因此可能会遇到此问题。情况会是这样吗？

另一种选择可能是使用基于BQ文件加载的写入方法而不是流插入。

不幸的是，基于存储写入API的写入模式目前也存在相同的限制。