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

使用 Datastream 和 Dataflow 流处理 流水线将数据复制到 BigQuery 中时，您可以 自定义 Dataflow 处理和合并更改的方式。根据自定义键对副本表进行分区、使用 JavaScript 用户定义的函数 (UDF) 转换事件，以及调整合并频率有助于您在数据新鲜度和查询与存储费用之间取得平衡。

根据用户定义的键对副本数据集进行分区

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 到 BigQuery 模板 重新创建 Dataflow 作业。

运行用户定义的函数来操纵事件数据

您可以使用 Datastream 到 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['is_deleted'] = messageJson['_metadata_deleted'];
   messageJson['_metadata_deleted'] = false;
    return JSON.stringify(messageJson);
 }

设置合并频率

使用 mergeFrequencyMinutes 参数设置 Datastream 到 BigQuery 模板 的合并频率。这是 BigQuery 中副本数据集内给定表的合并间隔分钟数。在回填历史数据时，我们建议您将合并频率保持在较低水平（12 或 24 小时），以控制费用。

例如，如果您将此参数的值设置为 10 分钟，则 Dataflow 将每 10 分钟运行一次使用该模板的作业。不过，该作业首次运行时，会有 5 分钟的延迟。在此示例中，如果作业在上午 9:14 运行，则第一次合并将在上午 9:29 进行（合并需要 10 分钟，延迟 5 分钟）。第二次合并将在上午 9:39 进行，之后的所有合并都将以 10 分钟为间隔进行（上午 9:49、上午 9:59、上午 10:09，依此类推）。

如果您将合并频率设置为 60 分钟，则该作业将在整点运行，首次运行该作业时会有 5 分钟的延迟。如果该作业计划在上午 10 点运行，则由于 5 分钟的延迟，它实际上会在上午 10:05 运行。之后的所有合并都将以 60 分钟为间隔进行（上午 11:05、下午 12:05、下午 1:05，依此类推）。

无论是为了控制费用还是出于其他原因，您可能无法以满足业务需求的频率执行合并。您可能无法获得最新鲜的数据。如需访问最新鲜的数据，请在 BigQuery 中暂存数据集和副本数据集的表之上创建一个视图，其中该视图会模拟合并。此视图会创建为一个逻辑表（适用于暂存数据集和副本数据集）。如果合并频率较低，并且您需要更快地访问数据，请使用该视图。