Use Datastream and Dataflow to stream data into BigQuery

In this page, you'll find best practices for using Datastream and Dataflow to stream data into BigQuery.

Partition replica datasets on user defined keys

The staging dataset in BigQuery is partitioned automatically. However, by default, the replica dataset isn't partitioned because the partition keys on the replica tables must be defined based on specific business logic, instead of being enforced by Datastream and Dataflow.

For each table in the replica dataset that needs partitioning:

1. Stop and drain the Dataflow job.

2. Use the SQL editor in BigQuery to run the following SQL script for each table in the replica dataset. For this example, the actor table in the datastream_cdc dataset has a last_update column that we want to set as our partition key. By running the script, you recreate the table with the correct partition key.

   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. Use the Datastream to BigQuery template to recreate a Dataflow job.

Run user-defined functions to manipulate event data

You can use the Datastream to BigQuery template to run a Javascript user-defined function. To do this, first, place a file containing the function in a specific location within Cloud Storage. Then, do the following:

• Use the javascriptTextTransformGcsPath parameter in the template to specify the location of the file in Cloud Storage that contains your user-defined function.
• Use the javascriptTextTransformFunctionName parameter to specify the name of the Javascript function that you want to call as your user-defined function.

For example, you can run a user-defined function to retain deleted records in the tables of the replica dataset within BigQuery. This process is known as a soft delete.

To achieve this, create a function that copies the value of the _metadata_deleted column into a new column named is_deleted, and then resets the _metadata_deleted column value to false. This causes the Dataflow job to ignore the delete events and retain the deleted records when updating the replica dataset in BigQuery.

Here's the sample code for this user-defined function:

/**
* 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);
 }

Set the merge frequency

Use the mergeFrequencyMinutes parameter of the Datastream to BigQuery template to set the merge frequency. This is the number of minutes between merges for a given table in the replica dataset in BigQuery. While historical data is being backfilled, we recommend that you keep the merge frequency low (12 or 24 hours) to keep costs under control.

For example, if you set the value of this parameter to 10 minutes, then Dataflow will run the job that uses the template every 10 minutes. However, the first time that the job runs, there'll be a 5-minute delay. For this example, if the job runs at 9:14 AM, then the first merge will occur at 9:29 AM (10 minutes for the merge and 5 minutes for the delay). The second merge will occur at 9:39 AM, and all subsequent merges will occur in 10-minute intervals (9:49 AM, 9:59AM, 10:09 AM, and so on).

If you set the merge frequency to 60 minutes, then the job will run on the hour, after a 5-minute delay for the initial running of the job. If the job is scheduled to run at 10 AM, then it will actually run at 10:05 AM because of the 5-minute delay. All subsequent merges will occur in 60-minute intervals (11:05 AM, 12:05 PM, 1:05 PM, and so on).

Whether as a result of controlling costs or for other reasons, you may not be able to perform a merge at a frequency that meets your business needs. You may not have the freshest data. To access the freshest data, create a view on top of the tables of the staging and replica datasets in BigQuery, where the view mimics the merge. This view is created as one logical table (for both the staging and replica datasets). If the merge frequency is low, and you need faster access to the data, then use the view.