BigQuery流缓冲区引发UPDATE/DELETE不支持问题及解决方案

BigQuery的流式插入（insertAll）会将数据先写入流式缓冲区（streaming buffer），该缓冲区的数据处于临时存储状态，尚未合并到表的持久化存储分区中。而UPDATE/DELETE这类DML操作仅支持操作已持久化到分区的数据，无法处理流式缓冲区里的内容，因此当表中存在未落盘的流式缓冲区数据时，执行全表删除就会触发该错误。

以下是几种适配你定时重写数据场景的可行方案：

方案1：改用批量加载替代流式插入 #

将原来的insertAll流式插入替换为批量加载方式，数据会直接写入持久化分区，不会产生流式缓冲区，后续即可正常执行DELETE操作。

调整后的代码示例：

BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
TableId tableId = TableId.of(DATASET, TABLE_NAME);

if (bigquery.getTable(tableId) != null) {
    // 执行全表删除
    QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder(
                    "DELETE FROM `" + DATASET + "." + TABLE_NAME + "` WHERE true;")
            .build();
    JobId jobId = JobId.of(UUID.randomUUID().toString());
    Job queryJob = bigquery.create(JobInfo.newBuilder(queryConfig).setJobId(jobId).build());
    queryJob = queryJob.waitFor();

    if (queryJob == null) {
        throw new RuntimeException("Job no longer exists");
    } else if (queryJob.getStatus().getError() != null) {
        throw new RuntimeException(queryJob.getStatus().getError().toString());
    }
} else {
    // 创建表
    Field id = Field.of("id", StandardSQLTypeName.INT64);
    Field name = Field.of("name", StandardSQLTypeName.STRING);
    Schema schema = Schema.of(id, name);
    TableDefinition tableDefinition = StandardTableDefinition.of(schema);
    TableInfo tableInfo = TableInfo.newBuilder(tableId, tableDefinition).build();
    bigquery.create(tableInfo);
}

// 批量写入数据（替代原insertAll逻辑）
List<TableRow> rows = new ArrayList<>();
for (Map.Entry<String, Object> entry : campaign.entrySet()) {
    TableRow row = new TableRow();
    row.set("id", entry.getKey()).set("name", entry.getValue());
    rows.add(row);
}

WriteChannelConfiguration writeConfig = WriteChannelConfiguration.newBuilder(tableId)
        .setFormatOptions(FormatOptions.json())
        .build();

try (WriteChannel writer = bigquery.writer(writeConfig)) {
    ByteArrayOutputStream out = new ByteArrayOutputStream();
    for (TableRow row : rows) {
        out.write(row.toPrettyString().getBytes(StandardCharsets.UTF_8));
        out.write('\n');
    }
    writer.write(ByteBuffer.wrap(out.toByteArray()));
}

方案2：创建临时表写入数据后替换原表 #

跳过删除原表数据的步骤，每次先将新数据写入临时表，再用临时表替换原表，彻底避开DML操作的限制。

代码示例：

BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
TableId targetTableId = TableId.of(DATASET, TABLE_NAME);
// 生成唯一临时表名
String tempTableName = TABLE_NAME + "_temp_" + UUID.randomUUID().toString().replace("-", "");
TableId tempTableId = TableId.of(DATASET, tempTableName);

// 创建临时表（复用原表结构）
Field id = Field.of("id", StandardSQLTypeName.INT64);
Field name = Field.of("name", StandardSQLTypeName.STRING);
Schema schema = Schema.of(id, name);
TableDefinition tableDefinition = StandardTableDefinition.of(schema);
TableInfo tempTableInfo = TableInfo.newBuilder(tempTableId, tableDefinition).build();
bigquery.create(tempTableInfo);

// 写入数据到临时表（可继续使用流式插入）
TableRow row = new TableRow();
for (Map.Entry<String, Object> entry : campaign.entrySet()) {
    row.set("id", entry.getKey()).set("name", entry.getValue());
    bigquery.insertAll(InsertAllRequest.newBuilder(tempTableId).addRow(row).build());
}

// 用临时表替换原表（WRITE_TRUNCATE会覆盖原表数据）
CopyJobConfiguration copyConfig = CopyJobConfiguration.newBuilder(
        tempTableId,
        targetTableId
).setWriteDisposition(WriteDisposition.WRITE_TRUNCATE).build();

JobId copyJobId = JobId.of(UUID.randomUUID().toString());
Job copyJob = bigquery.create(JobInfo.newBuilder(copyConfig).setJobId(copyJobId).build());
copyJob = copyJob.waitFor();

// 检查替换结果
if (copyJob == null || copyJob.getStatus().getError() != null) {
    String errorMsg = copyJob != null ? copyJob.getStatus().getError().toString() : "替换任务不存在";
    throw new RuntimeException("替换原表失败: " + errorMsg);
}

// 清理临时表
bigquery.delete(tempTableId);

方案3：使用分区表，删除指定分区而非全表 #

如果数据适合按时间分区（比如按写入小时划分），可以创建分区表，每次仅删除当前任务之前的旧分区，流式缓冲区的数据只会存在于最新分区，不会触发错误。

创建分区表的代码调整： #

// 创建按小时分区的表
Field id = Field.of("id", StandardSQLTypeName.INT64);
Field name = Field.of("name", StandardSQLTypeName.STRING);
Schema schema = Schema.of(id, name);

TimePartitioning timePartitioning = TimePartitioning.newBuilder(TimePartitioning.Type.HOUR).build();
TableDefinition tableDefinition = StandardTableDefinition.newBuilder(schema)
        .setTimePartitioning(timePartitioning)
        .build();

TableId tableId = TableId.of(DATASET, TABLE_NAME);
TableInfo tableInfo = TableInfo.newBuilder(tableId, tableDefinition).build();
bigquery.create(tableInfo);

替换全表删除的逻辑： #

// 删除当前时间前1小时的所有分区数据
String deleteQuery = String.format(
        "DELETE FROM `%s.%s` WHERE _PARTITIONTIME < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)",
        DATASET, TABLE_NAME
);

QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder(deleteQuery).build();
JobId jobId = JobId.of(UUID.randomUUID().toString());
Job queryJob = bigquery.create(JobInfo.newBuilder(queryConfig).setJobId(jobId).build());
queryJob = queryJob.waitFor();

// 错误检查逻辑同前...

方案4：等待缓冲区数据落盘（不推荐） #

流式缓冲区的数据通常会在10-15分钟内自动落盘到持久化分区，若任务允许延迟执行，可在DELETE前等待足够时长。但该方案会增加任务延迟，不适合你的每小时定时重写场景。

内容的提问来源于stack exchange，提问作者Mohamed Haydar