使用Dataflow的STREAMING_INSERT向BigQuery插入时请求大小超限问题

你的报错核心原因是Dataflow的BatchedStreamingWrite组件默认会将多条流记录打包成一个请求发送给BigQuery——即便单条记录只有1-2MB，多条打包后的总大小也会超过BigQuery STREAMING_INSERTS API的10MB请求限制。

以下是具体解决方案：

方案1：调整批处理的大小限制 #

通过设置批处理的最大字节数，确保每个请求的总大小不超过10MB（建议留1-2MB的余量，比如设置为8MB）。这种方式既保留了批处理的效率，又能避免触发大小限制。

修改后的代码示例：

.apply(
        "WriteSuccessfulRecords",
        BigQueryIO.writeTableRows().withAutoSharding()
                .withoutValidation()
                .withCreateDisposition(CreateDisposition.CREATE_NEVER)
                .withWriteDisposition(WriteDisposition.WRITE_APPEND)
                .withExtendedErrorInfo()
                .withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS)
                .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())
                .withBatchSizeBytes(8 * 1024 * 1024) // 设置单请求最大字节数为8MB
                .to(options.getOutputTableSpec()));

如果你的记录数量较少，也可以直接强制单条记录发送：

.withBatchSizeRows(1)

方案2：改用FILE_LOADS写入方式 #

如果场景对延迟要求不高，或者单条记录接近10MB，可以改用FILE_LOADS方式。这种方式会先将流数据写入GCS临时文件，再批量加载到BigQuery，没有单请求10MB的限制，适合大记录或高吞吐量场景。

修改后的代码示例：

.apply(
        "WriteSuccessfulRecords",
        BigQueryIO.writeTableRows().withAutoSharding()
                .withoutValidation()
                .withCreateDisposition(CreateDisposition.CREATE_NEVER)
                .withWriteDisposition(WriteDisposition.WRITE_APPEND)
                .withExtendedErrorInfo()
                .withMethod(BigQueryIO.Write.Method.FILE_LOADS)
                .withTempLocation("gs://your-gcs-bucket/temp-path") // 配置GCS临时路径
                .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())
                .to(options.getOutputTableSpec()));

注意事项 #

• BigQuery STREAMING_INSERTS的单条记录最大限制是10MB，你的单条记录1-2MB符合要求，问题仅出在批打包后的总大小。
• 若使用withBatchSizeRows(1)，需注意BigQuery的STREAMING_INSERTS QPS限制（默认每秒1000次请求），数据量过大时可能触发限流，此时优先考虑方案1或方案2。

内容的提问来源于stack exchange，提问作者Praful Janardhanan