如何更新 BigQuery 流缓冲区中的数据？

Question

我的工作流程

• 使用云功能将数据从 Pub/Sub 流式传输到 BigQuery。

• 数据在流缓冲区中保留 90 分钟，因此我无法执行更新语句。

• 我需要在那之前更新结果列。

  请帮忙。

我在“Pub/Sub”中接收数据，然后触发“云功能”，将数据插入“BigQuery”

这是代码：

const { BigQuery } = require('@google-cloud/bigquery');
const bigquery = new BigQuery();


exports.telemetryToBigQuery = (data, context) => {

  if (!data.data) {
    throw new Error('No telemetry data was provided!');
    return;
  }

  //Data comes in as base64
  console.log(`raw data: ${data.data}`);

  //Data gets decoded from base64 to string
  const dataDataDecode = Buffer.from(data.data, 'base64').toString();


var indexesSemicolons = [];

for (var i = 0; i < dataDataDecode.length; i++) {
    if (dataDataDecode[i] === ";") {
        indexesSemicolons.push(i);
    }
}

if (indexesSemicolons.length == 14) {

     const brand = dataDataDecode.slice(0, indexesSemicolons[0]);
     const model = dataDataDecode.slice(indexesSemicolons[0] + 1, indexesSemicolons[1]);
     const result = dataDataDecode.slice(indexesSemicolons[1] + 1, indexesSemicolons[2]);

    async function insertRowsAsStream() {
      // Inserts the JSON objects into my_dataset:my_table.

    
      const datasetId = 'put your dataset here';
      const tableId = 'put table id here';
      const rows = [
        {
          Brand: brand,
          Model: model,
          Result: result

        }
      ];

      // Insert data into a table
      await bigquery
        .dataset(datasetId)
        .table(tableId)
        .insert(rows);
      console.log(`Inserted ${rows.length} rows`);
    }
    insertRowsAsStream();
  } else {
    console.log("Invalid message");
    return;
  }
}

此数据在 BigQuery 流缓冲区中保留大约 90 分钟，但我需要执行一个更新查询来更改结果列。这是不允许的并且会导致错误

ApiError: UPDATE or DELETE statement over table pti-tag-copy.ContainerData2.voorinfo would affect rows in the streaming buffer, which is not supported at new ApiError

我需要一种方法在 90 分钟缓冲时间之前更新结果。你们能帮帮我吗？

我在线阅读了以下几页

BigQuery 流的生命周期

我阅读了以下问题的答案，我想我理解他在说什么，但我不知道如何执行它。

如果我是正确的，他是说将我的数据流式传输到临时表，然后从那里将其放入永久表中。

Stackoverflow DML 更新 bigQuery

Answer 1

是的，没错。当数据流式传输时，您不能使用 DML。解决方案是查询流缓冲区中的数据并将其转换到另一个表中。正如你所说，它可能是临时的，然后将它们放入永久的表中。

您还可以认为来自 PubSub 的流数据是原始数据，您希望保留它们，然后需要在另一个表中精炼数据。这也是一种常见的数据工程模式，具有不同的过滤和转换层，直到最终有用的数据（也称为数据集市）

Answer 2

回答你的问题。是的，它说您应该将数据流式传输到临时表并将其复制到另一个永久表，并且在原始表中您可以启用过期时间。这意味着该表将在过期时间过后被删除。

您可以更改过滤器，使其不包含当前流缓冲区中可能存在的数据。如果您在更新数据时使用分区表，则可以添加一个

WHERE

子句，其中时间戳的间隔为 40 到 90 分钟，例如：

WHERE Partitiontime < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 40 MINUTE).

Answer 3

我现在使用 BigQuery 创建作业方法。

示例位于此处

我直接将数据放入表中，这样我就不必等待流缓冲区 90 分钟。

Answer 4

我很高兴地宣布，公共预览版现在支持通过 BigQuery Storage Write API* 对最近的流数据进行变异 DML 语句（UPDATE、DELETE、MERGE）！在此处查看该功能以及如何将您的项目列入白名单：https://cloud.google.com/bigquery/docs/write-api#use_data_manipulation_language_dml_with_recently_streamed_data。

*此功能仅支持最近通过 BigQuery Storage Write API 传输的数据，而不支持旧版 insertAll 流式传输 API。