使用 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 user-defined function。如要這麼做，請先將含有函式的檔案放在 Cloud Storage 的特定位置。接著，按照下列步驟操作：

  • 使用範本中的 javascriptTextTransformGcsPath 參數，指定 Cloud Storage 中包含 user-defined function 的檔案位置。
  • 使用 javascriptTextTransformFunctionName 參數指定要呼叫的 Javascript 函式名稱，做為使用者定義函式。

舉例來說，您可以執行使用者定義函式，在 BigQuery 內保留副本資料表中的已刪除記錄。這項程序稱為「虛刪除」。

為此，請建立一個函式，將 _metadata_deleted 欄的值複製到名為 is_deleted 的新欄，然後將 _metadata_deleted 欄的值重設為 false。這會導致 Dataflow 工作在更新 BigQuery 中的副本資料集時，忽略刪除事件並保留已刪除的記錄。

以下是這個 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);
 }

設定合併頻率

使用 Datastream 到 BigQuery 範本的 mergeFrequencyMinutes 參數，設定合併頻率。這是指 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 點執行，則實際執行時間會是上午 10 點 5 分，因為延遲了 5 分鐘。之後的合併作業會以 60 分鐘為間隔 (上午 11:05、下午 12:05、下午 1:05 等)。

無論是為了控制成本或其他原因，您可能無法按照符合業務需求的方式執行合併。你可能沒有最新資料。如要存取最新資料，請在 BigQuery 中，於暫存和副本資料集的資料表上建立檢視區塊，該檢視區塊會模擬合併作業。這個檢視表會建立為一個邏輯資料表 (同時適用於暫存和副本資料集)。如果合併頻率較低，且您需要更快存取資料，請使用檢視畫面。