使用 Datastream 和 Dataflow 将数据流式插入到 BigQuery

在此页面中，您将找到有关如何使用 Datastream 和 Dataflow 将数据流式传输到 BigQuery 的最佳实践。

根据用户定义的键对复制的数据集进行分区

BigQuery 中的临时数据集会自动分区。不过，默认情况下，副本数据集不会进行分区，因为副本表的分区键必须根据特定的业务逻辑来定义，而不是由 Datastream 和 Dataflow 强制执行。

对于需要分区的副本数据集中的每个表：

  1. 停止并排空 Dataflow 作业

  2. 使用 BigQuery 中的 SQL 编辑器，针对复制数据集中的每个表运行以下 SQL 脚本。在此示例中，datastream_cdc 数据集中的 actor 表格具有一个 last_update 列，我们希望将其设置为分区键。运行该脚本后，您将使用正确的分区键重新创建表。

    create table '[BigQuery_PROJECT_ID].datastream_cdc.actor_new' partition by date(last_update)
as SELECT * FROM '[BigQuery_PROJECT_ID].datastream_cdc.actor'

drop table '[BigQuery_PROJECT_ID].datastream_cdc.actor'

alter table '[BigQuery_PROJECT_ID].datastream_cdc.actor_new' rename to 'actor'

  3. 使用 Datastream to BigQuery 模板重新创建 Dataflow 作业。

运行用户定义的函数来处理事件数据

您可以使用 Datastream to BigQuery 模板运行 JavaScript 用户定义的函数。为此，请先将包含该函数的文件放置在 Cloud Storage 中的特定位置。之后，执行以下操作：

  • 使用模板中的 javascriptTextTransformGcsPath 参数指定 Cloud Storage 中包含用户定义函数的文件所在的位置。
  • 使用 javascriptTextTransformFunctionName 参数指定要作为用户定义的函数调用的 JavaScript 函数的名称。

例如，您可以运行用户定义的函数，以在 BigQuery 中保留副本数据集的表中的已删除记录。此过程称为软删除。

为此，请创建一个函数，将 _metadata_deleted 列的值复制到名为 is_deleted 的新列中，然后将 _metadata_deleted 列的值重置为 false。这会导致 Dataflow 作业忽略删除事件，并在更新 BigQuery 中的副本数据集时保留已删除的记录。

以下是此用户定义的函数的示例代码：

/**
* This function changes the behavior of the Datastream to
* BigQuery template to allow soft deletes.
* @param {string} messageString from DatastreamIO data
* @return {string} same as an input message with an added property
*/
function transform(messageString) {
   // messageString is a JSON object as a string
   var messageJson = JSON.parse(messageString);
    // Moving the deleted flag to a new column will cause the pipeline to soft delete data.
   messageJson[