Bigtable change streams to BigQuery template

התבנית Bigtable change streams to BigQuery היא צינור להעברת נתונים (pipeline) שמעביר רשומות של שינויים בנתונים מ-Bigtable וכותב אותן בטבלאות ב-BigQuery באמצעות Dataflow.

Bigtable change stream מאפשר לכם להירשם לשינויים בנתונים על בסיס כל טבלה. כשנרשמים לסנכרון שינויים בזרמי נתונים בטבלה, חלים האילוצים הבאים:

• מוחזרים רק תאים ששונו ותיאורים של פעולות מחיקה.
• מוחזר רק הערך החדש של תא ששונה.

כשרושמים רשומות של שינויי נתונים ב-BigQuery, יכול להיות שהשורות יוכנסו לא לפי הסדר בהשוואה לסדר של חותמות הזמן המקוריות של ביצוע השינויים ב-Bigtable.

שורות בטבלת יומן השינויים שלא ניתן לכתוב ל-BigQuery בגלל שגיאה מתמשכת, מועברות באופן קבוע לספרייה של תור הודעות שלא עברו עיבוד (dead-letter queue) ב-Cloud Storage, לבדיקה על ידי משתמש או לעיבוד נוסף על ידי המשתמש.

אם הטבלה הנדרשת ב-BigQuery לא קיימת, צינור הנתונים יוצר אותה. אחרת, נעשה שימוש בטבלת BigQuery קיימת. הסכימה של טבלאות BigQuery קיימות צריכה לכלול את העמודות שבטבלה הבאה.

כל שורה חדשה ב-BigQuery כוללת רשומה אחת של שינוי נתונים שמוחזרת מזרם השינויים מהשורה התואמת בטבלת Bigtable.

סכימת טבלת הפלט ב-BigQuery

שם העמודה	סוג	Nullable	תיאור
row_key	STRING או BYTES	לא	מפתח השורה של השורה שהשתנתה. אם אפשרות הצינור writeRowkeyAsBytes מוגדרת כ-true, סוג העמודה חייב להיות BYTES. אחרת, משתמשים בסוג STRING.
mod_type	STRING	לא	סוג השינוי בשורה. משתמשים באחד מהערכים הבאים: SET_CELL, DELETE_CELLS או DELETE_FAMILY.
column_family	STRING	לא	קבוצת העמודות שהושפעה משינוי השורה.
column	STRING	כן	מגדיר העמודה שהושפע משינוי השורה. בסוג המוטציה DELETE_FAMILY, מגדירים את הערך NULL.
commit_timestamp	TIMESTAMP	לא	השעה שבה Bigtable מחיל את השינוי.
big_query_commit_timestamp	TIMESTAMP	כן	אופציונלי: מציין את השעה שבה BigQuery כותב את השורה לטבלת פלט. השדה לא יאוכלס אם שם העמודה מופיע בערך של אפשרות הצינור bigQueryChangelogTableFieldsToIgnore.
timestamp	TIMESTAMP או INT64	כן	ערך חותמת הזמן של התא שהמוטציה משפיעה עליו. אם אפשרות הצינור writeNumericTimestamps מוגדרת כ-true, סוג העמודה חייב להיות INT64. אחרת, משתמשים בסוג TIMESTAMP. לסוגי המוטציות DELETE_CELLS ו-DELETE_FAMILY, הערך שמוגדר הוא NULL.
timestamp_from	TIMESTAMP או INT64	כן	מתאר את ההתחלה של מרווח הזמן של חותמות הזמן, כולל, לכל התאים שנמחקו על ידי המוטציה DELETE_CELLS. לסוגים אחרים של מוטציות, מגדירים את הערך NULL.
timestamp_to	TIMESTAMP או INT64	כן	מתאר את הסוף הבלעדי של מרווח חותמות הזמן של כל התאים שנמחקו על ידי המוטציה DELETE_CELLS. לסוגים אחרים של מוטציות, מגדירים את הערך NULL.
is_gc	BOOL	לא	אופציונלי: אם המוטציה מופעלת על ידי מדיניות איסוף נתונים מיותרים, צריך להגדיר את הערך true. בכל שאר המקרים, מגדירים את הערך false. השדה לא מאוכלס כששם העמודה מופיע בערך האפשרות של צינור bigQueryChangelogTableFieldsToIgnore.
source_instance	STRING	לא	אופציונלי: תיאור של השם של מופע Bigtable שממנו מגיעה המוטציה. השדה לא מאוכלס כששם העמודה מופיע בערך האפשרות של צינור bigQueryChangelogTableFieldsToIgnore.
source_cluster	STRING	לא	אופציונלי: תיאור של שם אשכול Bigtable שממנו מגיע השינוי. השדה לא מאוכלס כששם העמודה מופיע בערך האפשרות של צינור bigQueryChangelogTableFieldsToIgnore.
source_table	STRING	לא	אופציונלי: תיאור של שם טבלת Bigtable שהמוטציה חלה עליה. הערך בעמודה הזו יכול להיות שימושי אם כמה טבלאות Bigtable מעבירות שינויים בסטרימינג לאותה טבלה ב-BigQuery. השדה לא מאוכלס כששם העמודה מופיע בערך האפשרות של צינור bigQueryChangelogTableFieldsToIgnore.
tiebreaker	INT64	לא	אופציונלי: כששני שינויים נרשמים בו-זמנית על ידי אשכולות שונים של Bigtable, השינוי עם הערך tiebreaker הגבוה ביותר מוחל על טבלת המקור. מוטציות עם ערכים נמוכים יותר של tiebreaker נפסלות. השדה לא מאוכלס כששם העמודה מופיע בערך האפשרות של צינור bigQueryChangelogTableFieldsToIgnore.
value	STRING או BYTES	כן	הערך החדש שהוגדר על ידי השינוי. אם אפשרות הצינור writeValuesAsBytes מוגדרת כ-true, סוג העמודה חייב להיות BYTES. אחרת, משתמשים בסוג STRING. הערך מוגדר למוטציות של SET_CELL. בסוגים אחרים של מוטציות, הערך שמוגדר הוא NULL.

הדרישות לגבי צינורות עיבוד נתונים

• מופע המקור של Bigtable שצוין.
• טבלת המקור שצוינה ב-Bigtable. צריך להפעיל את סנכרון שינויים בזרמי נתונים בטבלה.
• פרופיל האפליקציה שצוין ב-Bigtable.
• מערך הנתונים של היעד שצוין ב-BigQuery.

פרמטרים של תבניות

פרמטרים נדרשים

• ‫bigQueryDataset: השם של מערך הנתונים של טבלת היעד ב-BigQuery.
• ‫bigtableChangeStreamAppProfile: מזהה פרופיל האפליקציה של Bigtable. פרופיל האפליקציה צריך להשתמש בניתוב של אשכול יחיד ולאפשר עסקאות של שורה אחת.
• ‫bigtableReadInstanceId: מזהה מכונת Bigtable של המקור.
• ‫bigtableReadTableId: מזהה טבלת Bigtable של המקור.

פרמטרים אופציונליים

• ‫writeRowkeyAsBytes: האם לכתוב את מפתחות השורות כ-BigQuery BYTES. אם הערך הוא true, מפתחות השורות נכתבים בעמודה BYTES. אחרת, מפתחות השורות נכתבים בעמודה STRING. ברירת המחדל היא false.
• ‫writeValuesAsBytes: אם הערך הוא true, הערכים נכתבים בעמודה מסוג BYTES, אחרת בעמודה מסוג STRING . ברירת המחדל: false.
• ‫writeNumericTimestamps: האם לכתוב את חותמת הזמן של Bigtable כ-INT64 ב-BigQuery. כשמגדירים את הערך true, הערכים נכתבים בעמודה INT64. אחרת, הערכים נכתבים בעמודה TIMESTAMP. העמודות שהושפעו: timestamp,‏ timestamp_from וtimestamp_to. ברירת המחדל היא false. אם הערך הוא true, הזמן נמדד במיקרו-שניות מאז ראשית זמן יוניקס (1 בינואר 1970 ב-UTC).
• ‫bigQueryProjectId: מזהה הפרויקט של מערך הנתונים ב-BigQuery. ברירת המחדל היא הפרויקט של משימת Dataflow.
• ‫bigQueryChangelogTableName: שם טבלת היעד ב-BigQuery. אם לא מציינים ערך, המערכת משתמשת בערך bigtableReadTableId + "_changelog". ברירת המחדל היא ריק.
• ‫bigQueryChangelogTablePartitionGranularity: מציין את רמת הפירוט לחלוקת טבלת יומן השינויים למחיצות. אם מגדירים את האפשרות הזו, הטבלה מחולקת למחיצות. משתמשים באחד מהערכים הנתמכים הבאים: HOUR,‏ DAY,‏ MONTH או YEAR. כברירת מחדל, הטבלה לא מחולקת למחיצות.
• ‫bigQueryChangelogTablePartitionExpirationMs: הגדרת זמן התפוגה של מחיצות בטבלת יומן השינויים, באלפיות השנייה. אם הערך הוא true, מחיצות שגילן גדול ממספר אלפיות השנייה שצוין יימחקו. כברירת מחדל, לא מוגדר תאריך תפוגה.
• ‫bigQueryChangelogTableFieldsToIgnore: רשימה מופרדת בפסיקים של עמודות ביומן השינויים, שאם מציינים אותן, הן לא נוצרות ולא מאוכלסות. משתמשים באחד מהערכים הנתמכים הבאים: is_gc,‏ source_instance,‏ source_cluster,‏ source_table,‏ tiebreaker או big_query_commit_timestamp. כברירת מחדל, כל העמודות מאוכלסות.
• ‫dlqDirectory: הספרייה שבה יש להשתמש לתור של הודעות שלא ניתן להעביר. רשומות שלא עוברות עיבוד מאוחסנות בספרייה הזו. ברירת המחדל היא ספרייה במיקום הזמני של משימת Dataflow. ברוב המקרים אפשר להשתמש בנתיב ברירת המחדל.
• ‫bigtableChangeStreamMetadataInstanceId: מזהה מופע של מטא-נתונים של Bigtable סנכרון שינויים בזרמי נתונים. ברירת המחדל היא ריק.
• ‫bigtableChangeStreamMetadataTableTableId: המזהה של טבלת המטא-נתונים של מחבר Bigtable סנכרון שינויים בזרמי נתונים. אם לא מספקים את המידע הזה, המערכת יוצרת באופן אוטומטי טבלת מטא-נתונים של מחבר Bigtable לסנכרון שינויים בזרמי נתונים במהלך ההרצה של צינור הנתונים. ברירת המחדל היא ריק.
• ‫bigtableChangeStreamCharset: שם charset של סנכרון שינויים בזרמי נתונים של Bigtable. ברירת המחדל היא UTF-8.
• ‫bigtableChangeStreamStartTimestamp: חותמת הזמן (https://tools.ietf.org/html/rfc3339) שמשמשת לקריאת סנכרון שינויים בזרמי נתונים, כולל חותמת הזמן עצמה. לדוגמה, 2022-05-05T07:59:59Z. ברירת המחדל היא חותמת הזמן של שעת ההתחלה של הצינור.
• ‫bigtableChangeStreamIgnoreColumnFamilies: רשימה מופרדת בפסיקים של שינויים בשמות של משפחות עמודות שצריך להתעלם מהם. ברירת המחדל היא ריק.
• ‫bigtableChangeStreamIgnoreColumns: רשימה מופרדת בפסיקים של שינויים בשמות עמודות שצריך להתעלם מהם. דוגמה: "cf1:col1,cf2:col2". ברירת המחדל היא ריק.
• ‫bigtableChangeStreamName: שם ייחודי של צינור הנתונים של הלקוח. מאפשרת לכם להמשיך את העיבוד מהנקודה שבה צינור קודם שהופעל נעצר. ברירת המחדל היא שם שנוצר באופן אוטומטי. הערך שבו נעשה שימוש מופיע ביומני המשימות של Dataflow.
• ‫bigtableChangeStreamResume: אם הערך הוא true, צינור חדש ימשיך את העיבוד מהנקודה שבה צינור שפעל בעבר עם אותו ערך bigtableChangeStreamName הפסיק. אם צינור העיבוד עם הערך bigtableChangeStreamName שצוין לא הופעל אף פעם, לא יופעל צינור עיבוד חדש. כשמגדירים את האפשרות false, מתחיל צינור חדש. אם צינור עם אותו ערך bigtableChangeStreamName כבר הופעל עבור המקור הנתון, לא יופעל צינור חדש. ברירת המחדל היא false.
• ‫bigtableReadChangeStreamTimeoutMs: משך הזמן הקצוב לתפוגה של בקשות Bigtable ReadChangeStream באלפיות השנייה.
• ‫bigtableReadProjectId: מזהה הפרויקט ב-Bigtable. ברירת המחדל היא הפרויקט של משימת Dataflow.

הפעלת התבנית

המסוף

1. עוברים לדף Dataflow Create job from template (יצירת משימה מתבנית).
כניסה לדף Create job from template
2. בשדה שם המשימה, מזינים שם ייחודי למשימה.
3. אופציונלי: בשדה Regional endpoint (נקודת קצה אזורית), בוחרים ערך מהתפריט הנפתח. אזור ברירת המחדל הוא us-central1.

   רשימת האזורים שבהם אפשר להריץ משימת Dataflow מופיעה במאמר מיקומי Dataflow.

4. בתפריט הנפתח Dataflow template (תבנית של העברת נתונים), בוחרים באפשרות the Bigtable change streams to BigQuery template.
5. בשדות הפרמטרים שמופיעים, מזינים את ערכי הפרמטרים.
6. לוחצים על הפעלת העבודה.

gcloud

הערה: כדי להשתמש ב-Google Cloud CLI להפעלת תבניות flex, צריך להתקין את Google Cloud CLI בגרסה 284.0.0, או בגרסה מאוחרת יותר.

במעטפת או בטרמינל, מריצים את התבנית:

gcloud dataflow flex-template run JOB_NAME \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_BigQuery \
    --parameters \
bigtableReadInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableReadTableId=BIGTABLE_TABLE_ID,\
bigtableChangeStreamAppProfile=BIGTABLE_APPLICATION_PROFILE_ID,\
bigQueryDataset=BIGQUERY_DESTINATION_DATASET

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שגרסת התבניות העדכנית תתעדכן עם שינויים שעלולים לשבור את התאימות לאחור. כדי למנוע מהשינויים האלה לשבש את תהליכי העבודה בסביבת הייצור, צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר בסביבות הייצור.
• ‫REGION_NAME: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫BIGTABLE_INSTANCE_ID: מזהה מופע Bigtable.
• ‫BIGTABLE_TABLE_ID: מזהה הטבלה ב-Bigtable.
• ‫BIGTABLE_APPLICATION_PROFILE_ID: מזהה פרופיל האפליקציה של Bigtable.
• ‫BIGQUERY_DESTINATION_DATASET: השם של מערך הנתונים של היעד ב-BigQuery

API

כדי להריץ את התבנית באמצעות API בארכיטקטורת REST, שולחים בקשת HTTP POST. מידע נוסף על ה-API ועל היקפי ההרשאות שלו זמין במאמר projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
  "launch_parameter": {
    "jobName": "JOB_NAME",
    "containerSpecGcsPath": "gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_BigQuery",
    "parameters": {
        "bigtableReadInstanceId": "BIGTABLE_INSTANCE_ID",
        "bigtableReadTableId": "BIGTABLE_TABLE_ID",
        "bigtableChangeStreamAppProfile": "BIGTABLE_APPLICATION_PROFILE_ID",
        "bigQueryDataset": "BIGQUERY_DESTINATION_DATASET"
    }
  }
}

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שגרסת התבניות העדכנית תתעדכן עם שינויים שעלולים לשבור את התאימות לאחור. כדי למנוע מהשינויים האלה לשבש את תהליכי העבודה בסביבת הייצור, צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר בסביבות הייצור.
• ‫LOCATION: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫BIGTABLE_INSTANCE_ID: מזהה מופע Bigtable.
• ‫BIGTABLE_TABLE_ID: מזהה הטבלה ב-Bigtable.
• ‫BIGTABLE_APPLICATION_PROFILE_ID: מזהה פרופיל האפליקציה של Bigtable.
• ‫BIGQUERY_DESTINATION_DATASET: השם של מערך הנתונים של היעד ב-BigQuery

קוד המקור של התבנית

Java

/*
 * Copyright (C) 2023 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery;

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.Timestamp;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.DatasetId;
import com.google.cloud.bigtable.data.v2.models.ChangeStreamMutation;
import com.google.cloud.bigtable.data.v2.models.DeleteCells;
import com.google.cloud.bigtable.data.v2.models.DeleteFamily;
import com.google.cloud.bigtable.data.v2.models.Entry;
import com.google.cloud.bigtable.data.v2.models.SetCell;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions.ReadChangeStreamOptions;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions.ReadOptions;
import com.google.cloud.teleport.v2.bigtable.utils.UnsupportedEntryException;
import com.google.cloud.teleport.v2.cdc.dlq.DeadLetterQueueManager;
import com.google.cloud.teleport.v2.options.BigtableChangeStreamToBigQueryOptions;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.BigQueryDestination;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.Mod;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.ModType;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.schemautils.BigQueryUtils;
import com.google.cloud.teleport.v2.transforms.DLQWriteTransform;
import com.google.cloud.teleport.v2.utils.BigtableSource;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.InsertRetryPolicy;
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO.ExistingPipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Values;
import org.apache.beam.sdk.values.PCollection;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This pipeline ingests {@link ChangeStreamMutation} from Bigtable change stream. The {@link
 * ChangeStreamMutation} is then broken into {@link Mod}, which converted into {@link TableRow} and
 * inserted into BigQuery table.
 */
@Template(
    name = "Bigtable_Change_Streams_to_BigQuery",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Bigtable Change Streams to BigQuery",
    description =
        "Streaming pipeline. Streams Bigtable data change records and writes them into BigQuery using Dataflow Runner V2.",
    optionsClass = BigtableChangeStreamToBigQueryOptions.class,
    optionsOrder = {
      BigtableChangeStreamToBigQueryOptions.class,
      ReadChangeStreamOptions.class,
      ReadOptions.class
    },
    skipOptions = {
      "bigtableReadAppProfile",
      "bigtableAdditionalRetryCodes",
      "bigtableRpcAttemptTimeoutMs",
      "bigtableRpcTimeoutMs"
    },
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-bigtable-change-streams-to-bigquery",
    flexContainerName = "googlecloud-to-googlecloud",
    contactInformation = "https://cloud.google.com/support",
    streaming = true)
public final class BigtableChangeStreamsToBigQuery {
  private static final Logger LOG = LoggerFactory.getLogger(BigtableChangeStreamsToBigQuery.class);

  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

  /**
   * Main entry point for executing the pipeline.
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) {
    LOG.info("Starting to replicate change records from Cloud Bigtable change streams to BigQuery");

    BigtableChangeStreamToBigQueryOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(BigtableChangeStreamToBigQueryOptions.class);

    run(options);
  }

  private static void setOptions(BigtableChangeStreamToBigQueryOptions options) {
    options.setStreaming(true);
    options.setEnableStreamingEngine(true);

    // Add use_runner_v2 to the experiments option, since change streams connector is only supported
    // on Dataflow runner v2.
    List<String> experiments = options.getExperiments();
    if (experiments == null) {
      experiments = new ArrayList<>();
    }
    boolean hasUseRunnerV2 = false;
    for (String experiment : experiments) {
      if (experiment.equalsIgnoreCase(USE_RUNNER_V2_EXPERIMENT)) {
        hasUseRunnerV2 = true;
        break;
      }
    }
    if (!hasUseRunnerV2) {
      experiments.add(USE_RUNNER_V2_EXPERIMENT);
    }
    options.setExperiments(experiments);
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  public static PipelineResult run(BigtableChangeStreamToBigQueryOptions options) {
    setOptions(options);

    String changelogTableName = getBigQueryChangelogTableName(options);
    String bigtableProject = getBigtableProjectId(options);
    String bigQueryProject = getBigQueryProjectId(options);
    String bigQueryDataset = options.getBigQueryDataset();

    // If dataset doesn't exist and not checked, pipeline will start failing only after it sees the
    // first change from Cloud Bigtable. BigQueryIO can create table if it doesn't exist, but it
    // cannot create a dataset
    validateBigQueryDatasetExists(bigQueryProject, bigQueryDataset);

    // Retrieve and parse the startTimestamp
    Instant startTimestamp =
        options.getBigtableChangeStreamStartTimestamp().isEmpty()
            ? Instant.now()
            : toInstant(Timestamp.parseTimestamp(options.getBigtableChangeStreamStartTimestamp()));

    BigtableSource sourceInfo =
        new BigtableSource(
            options.getBigtableReadInstanceId(),
            options.getBigtableReadTableId(),
            getBigtableCharset(options),
            options.getBigtableChangeStreamIgnoreColumnFamilies(),
            options.getBigtableChangeStreamIgnoreColumns(),
            startTimestamp);

    BigQueryDestination destinationInfo =
        new BigQueryDestination(
            bigQueryProject,
            bigQueryDataset,
            changelogTableName,
            options.getWriteRowkeyAsBytes(),
            options.getWriteValuesAsBytes(),
            options.getWriteNumericTimestamps(),
            options.getBigQueryChangelogTablePartitionGranularity(),
            options.getBigQueryChangelogTablePartitionExpirationMs(),
            options.getBigQueryChangelogTableFieldsToIgnore());

    BigQueryUtils bigQuery = new BigQueryUtils(sourceInfo, destinationInfo);

    Pipeline pipeline = Pipeline.create(options);
    DeadLetterQueueManager dlqManager = buildDlqManager(options);

    BigtableIO.ReadChangeStream readChangeStream =
        BigtableIO.readChangeStream()
            .withChangeStreamName(options.getBigtableChangeStreamName())
            .withExistingPipelineOptions(
                options.getBigtableChangeStreamResume()
                    ? ExistingPipelineOptions.RESUME_OR_FAIL
                    : ExistingPipelineOptions.FAIL_IF_EXISTS)
            .withProjectId(bigtableProject)
            .withMetadataTableInstanceId(options.getBigtableChangeStreamMetadataInstanceId())
            .withInstanceId(options.getBigtableReadInstanceId())
            .withTableId(options.getBigtableReadTableId())
            .withAppProfileId(options.getBigtableChangeStreamAppProfile())
            .withStartTime(startTimestamp);

    if (!StringUtils.isBlank(options.getBigtableChangeStreamMetadataTableTableId())) {
      readChangeStream =
          readChangeStream.withMetadataTableTableId(
              options.getBigtableChangeStreamMetadataTableTableId());
    }

    PCollection<ChangeStreamMutation> dataChangeRecord =
        pipeline
            .apply("Read from Cloud Bigtable Change Streams", readChangeStream)
            .apply(Values.create());

    PCollection<TableRow> changeStreamMutationToTableRow =
        dataChangeRecord.apply(
            "ChangeStreamMutation To TableRow",
            ParDo.of(new ChangeStreamMutationToTableRowFn(sourceInfo, bigQuery)));

    Write<TableRow> bigQueryWrite =
        BigQueryIO.<TableRow>write()
            .to(destinationInfo.getBigQueryTableReference())
            .withSchema(bigQuery.getDestinationTableSchema())
            .withFormatFunction(element -> element)
            .withFormatRecordOnFailureFunction(element -> element)
            .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
            .withWriteDisposition(WriteDisposition.WRITE_APPEND)
            .withExtendedErrorInfo()
            .withMethod(Write.Method.STORAGE_API_AT_LEAST_ONCE)
            .withNumStorageWriteApiStreams(0)
            .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors());

    if (destinationInfo.isPartitioned()) {
      bigQueryWrite = bigQueryWrite.withTimePartitioning(bigQuery.getTimePartitioning());
    }

    // Unfortunately, due to https://github.com/apache/beam/issues/24090, it is no longer possible
    // to pass metadata via fake columns when writing to BigQuery. Previously we'd pass something
    // like retry count and then format it out before writing, but BQ would return original object
    // which would allow us to increment retry count and store it to DLQ with incremented number.
    // Because WRITE API doesn't allow access to original object, all metadata values are stripped
    // and we can only rely on retry policy and put all other persistently failing rows to DLQ as
    // a non-retriable severe failure.
    //
    // Since we're not going to be retrying such failures, we'll not use any reading from DLQ
    // capability.

    WriteResult writeResult =
        changeStreamMutationToTableRow.apply("Write To BigQuery", bigQueryWrite);

    writeResult
        .getFailedStorageApiInserts()
        .apply(
            "Failed Mod JSON During BigQuery Writes",
            MapElements.via(new BigQueryDeadLetterQueueSanitizer()))
        .apply(
            "Write rejected TableRow JSON To DLQ",
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqManager.getSevereDlqDirectory() + "YYYY/MM/dd/HH/mm/")
                .withTmpDirectory(dlqManager.getSevereDlqDirectory() + "tmp/")
                .setIncludePaneInfo(true)
                .build());

    return pipeline.run();
  }

  private static void validateBigQueryDatasetExists(
      String bigQueryProject, String bigQueryDataset) {
    BigQueryOptions options = BigQueryOptions.newBuilder().build();
    options.setThrowNotFound(true);

    BigQuery bigQuery = options.getService();
    bigQuery.getDataset(DatasetId.of(bigQueryProject, bigQueryDataset));
  }

  private static Instant toInstant(Timestamp timestamp) {
    if (timestamp == null) {
      return null;
    } else {
      return Instant.ofEpochMilli(timestamp.getSeconds() * 1000 + timestamp.getNanos() / 1000000);
    }
  }

  private static DeadLetterQueueManager buildDlqManager(
      BigtableChangeStreamToBigQueryOptions options) {
    String tempLocation =
        options.as(DataflowPipelineOptions.class).getTempLocation().endsWith("/")
            ? options.as(DataflowPipelineOptions.class).getTempLocation()
            : options.as(DataflowPipelineOptions.class).getTempLocation() + "/";
    String dlqDirectory =
        options.getDlqDirectory().isEmpty() ? tempLocation + "dlq/" : options.getDlqDirectory();

    LOG.info("Dead letter queue directory: {}", dlqDirectory);
    return DeadLetterQueueManager.create(dlqDirectory, 1);
  }

  private static String getBigtableCharset(BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigtableChangeStreamCharset())
        ? "UTF-8"
        : options.getBigtableChangeStreamCharset();
  }

  private static String getBigtableProjectId(BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigtableReadProjectId())
        ? options.getProject()
        : options.getBigtableReadProjectId();
  }

  private static String getBigQueryChangelogTableName(
      BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigQueryChangelogTableName())
        ? options.getBigtableReadTableId() + "_changelog"
        : options.getBigQueryChangelogTableName();
  }

  private static String getBigQueryProjectId(BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigQueryProjectId())
        ? options.getProject()
        : options.getBigQueryProjectId();
  }

  /**
   * DoFn that converts a {@link ChangeStreamMutation} to multiple {@link Mod} in serialized JSON
   * format.
   */
  static class ChangeStreamMutationToTableRowFn extends DoFn<ChangeStreamMutation, TableRow> {
    private final BigtableSource sourceInfo;
    private final BigQueryUtils bigQuery;

    ChangeStreamMutationToTableRowFn(BigtableSource source, BigQueryUtils bigQuery) {
      this.sourceInfo = source;
      this.bigQuery = bigQuery;
    }

    @ProcessElement
    public void process(@Element ChangeStreamMutation input, OutputReceiver<TableRow> receiver)
        throws Exception {
      for (Entry entry : input.getEntries()) {
        ModType modType = getModType(entry);

        Mod mod = null;
        switch (modType) {
          case SET_CELL:
            mod = new Mod(sourceInfo, input, (SetCell) entry);
            break;
          case DELETE_CELLS:
            mod = new Mod(sourceInfo, input, (DeleteCells) entry);
            break;
          case DELETE_FAMILY:
            mod = new Mod(sourceInfo, input, (DeleteFamily) entry);
            break;
          default:
          case UNKNOWN:
            throw new UnsupportedEntryException(
                "Cloud Bigtable change stream entry of type "
                    + entry.getClass().getName()
                    + " is not supported. The entry was put into a dead letter queue directory. "
                    + "Please update your Dataflow template with the latest template version");
        }

        TableRow tableRow = new TableRow();
        if (bigQuery.setTableRowFields(mod, tableRow)) {
          receiver.output(tableRow);
        }
      }
    }

    private ModType getModType(Entry entry) {
      if (entry instanceof SetCell) {
        return ModType.SET_CELL;
      } else if (entry instanceof DeleteCells) {
        return ModType.DELETE_CELLS;
      } else if (entry instanceof DeleteFamily) {
        return ModType.DELETE_FAMILY;
      }
      return ModType.UNKNOWN;
    }
  }
}

המאמרים הבאים

• מידע נוסף על תבניות Dataflow
• רשימת התבניות ש-Google סיפקה