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: הספרייה שבה ישמש תור של הודעות שלא ניתן להעביר (DLQ). רשומות שלא ניתן לעבד נשמרות בספרייה הזו. ברירת המחדל היא ספרייה במיקום הזמני של משימת Dataflow. ברוב המקרים, אפשר להשתמש בנתיב ברירת המחדל.
• ‫bigtableChangeStreamMetadataInstanceId: מזהה מופע של מטא-נתונים של Bigtable change streams. ברירת המחדל היא ריק.
• ‫bigtableChangeStreamMetadataTableTableId: המזהה של טבלת המטא-נתונים של מחבר Bigtable change streams. אם לא מספקים את המידע הזה, נוצרת באופן אוטומטי טבלת מטא-נתונים של מחבר Bigtable לשינויים בזמן אמת במהלך ההפעלה של צינור הנתונים. ברירת המחדל היא ריק.
• ‫bigtableChangeStreamCharset: שם ערכת התווים של סנכרון שינויים בזרמי נתונים של 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, בוחרים בתבנית Bigtable change streams to 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-REGION_NAME/VERSION/flex/ \
    --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/",
    "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 סיפקה.