Bigtable change streams to BigQuery template

התבנית Bigtable change streams to BigQuery היא פייפליין של סנכרון שינויים בזרמי נתונים, שמעביר רשומות של שינויים בנתונים ב-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 change streams. ברירת המחדל היא ריקה.
• ‫bigtableChangeStreamMetadataTableTableId: המזהה של טבלת המטא-נתונים של מחבר Bigtable change streams. אם לא מספקים את המידע הזה, נוצרת באופן אוטומטי טבלת מטא-נתונים של מחבר 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. עוברים לדף Create job from template (יצירת משימה מתבנית) ב-Dataflow.
כניסה לדף Create job from template
2. בשדה שם המשימה, מזינים שם ייחודי למשימה.
3. אופציונלי: בשדה Regional endpoint (נקודת קצה אזורית), בוחרים ערך מהתפריט הנפתח. אזור ברירת המחדל הוא us-central1.

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

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

gcloud

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

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

gcloud dataflow flex-template run JOB_NAME \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates/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: מזהה הפרויקט שבו רוצים להריץ את משימת Google Cloud Dataflow
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שרוצים להשתמש בה

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

  • latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית. אפשר למצוא את הגרסה הזו בתיקיית ההורה המתאימה עם התאריך בדלי – gs://dataflow-templates/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫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/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: מזהה הפרויקט שבו רוצים להריץ את משימת Google Cloud Dataflow
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שרוצים להשתמש בה

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

  • latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית. אפשר למצוא את הגרסה הזו בתיקיית ההורה המתאימה עם התאריך בדלי – gs://dataflow-templates/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫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 סיפקה.