תבנית של סנכרון שינויים בזרמי נתונים מ-Bigtable ל-Pub/Sub

התבנית Bigtable change streams to Pub/Sub היא צינור עיבוד נתונים בסטרימינג שמעביר בסטרימינג רשומות של שינויים בנתונים ב-Bigtable ומפרסם אותן בנושא ב-Pub/Sub באמצעות Dataflow.

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

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

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

רשומות של שינויים בנתונים ב-Bigtable שלא ניתן לפרסם בנושאים ב-Pub/Sub מועברות באופן זמני לספרייה של תור הודעות שלא ניתן לעיבוד (dead-letter queue) ב-Cloud Storage. אחרי מספר הניסיונות המקסימלי הכושל, הרשומות האלה מועברות לצמיתות לאותו מדריך של תור הודעות שלא ניתן למסור לבדיקה אנושית או לעיבוד נוסף על ידי המשתמש.

הצינור דורש שנושא היעד ב-Pub/Sub יתקיים. יכול להיות שהנושא של היעד מוגדר לאימות הודעות באמצעות סכימה. אם נושא Pub/Sub מציין סכימה, הצינור יתחיל לפעול רק אם הסכימה תקפה. בהתאם לסוג הסכימה, משתמשים באחת מהגדרות הסכימה הבאות לנושא היעד:

מאגרי אחסון לפרוטוקולים

syntax = "proto2";

package com.google.cloud.teleport.bigtable;

option java_outer_classname = "ChangeLogEntryProto";

message ChangelogEntryProto{
  required bytes rowKey = 1;
  enum ModType {
    SET_CELL = 0;
    DELETE_FAMILY = 1;
    DELETE_CELLS = 2;
    UNKNOWN = 3;
  }
  required ModType modType = 2;
  required bool isGC = 3;
  required int32 tieBreaker = 4;
  required int64 commitTimestamp = 5;
  required string columnFamily = 6;
  optional bytes column = 7;
  optional int64 timestamp = 8;
  optional int64 timestampFrom = 9;
  optional int64 timestampTo = 10;
  optional bytes value = 11;
  required string sourceInstance = 12;
  required string sourceCluster = 13;
  required string sourceTable = 14;
}
  

Avro

{
    "name" : "ChangelogEntryMessage",
    "type" : "record",
    "namespace" : "com.google.cloud.teleport.bigtable",
    "fields" : [
      { "name" : "rowKey", "type" : "bytes"},
      {
        "name" : "modType",
        "type" : {
          "name": "ModType",
          "type": "enum",
          "symbols": ["SET_CELL", "DELETE_FAMILY", "DELETE_CELLS", "UNKNOWN"]}
      },
      { "name": "isGC", "type": "boolean" },
      { "name": "tieBreaker", "type": "int"},
      { "name": "columnFamily", "type": "string"},
      { "name": "commitTimestamp", "type" : "long"},
      { "name" : "sourceInstance", "type" : "string"},
      { "name" : "sourceCluster", "type" : "string"},
      { "name" : "sourceTable", "type" : "string"},
      { "name": "column", "type" : ["null", "bytes"]},
      { "name": "timestamp", "type" : ["null", "long"]},
      { "name": "timestampFrom", "type" : ["null", "long"]},
      { "name": "timestampTo", "type" : ["null", "long"]},
      { "name" : "value", "type" : ["null", "bytes"]}
   ]
}
    

JSON

משתמשים בסכימת Protobuf הבאה עם קידוד הודעות JSON:

syntax = "proto2";

package com.google.cloud.teleport.bigtable;

option java_outer_classname = "ChangelogEntryMessageText";

message ChangelogEntryText{
  required string rowKey = 1;
  enum ModType {
    SET_CELL = 0;
    DELETE_FAMILY = 1;
    DELETE_CELLS = 2;
    UNKNOWN = 3;
  }
  required ModType modType = 2;
  required bool isGC = 3;
  required int32 tieBreaker = 4;
  required int64 commitTimestamp = 5;
  required string columnFamily = 6;
  optional string column = 7;
  optional int64 timestamp = 8;
  optional int64 timestampFrom = 9;
  optional int64 timestampTo = 10;
  optional string value = 11;
  required string sourceInstance = 12;
  required string sourceCluster = 13;
  required string sourceTable = 14;
}
    

כל הודעת Pub/Sub חדשה כוללת רשומה אחת מרשומה של שינוי נתונים שמוחזרת מזרם השינויים מהשורה התואמת בטבלת Bigtable. התבנית Pub/Sub משטחת את הרשומות בכל רשומה של שינוי נתונים לשינויים ברמת התא הבודד.

תיאור הודעת הפלט של Pub/Sub

שם השדה	תיאור
rowKey	מפתח השורה של השורה שהשתנתה. הנתונים מגיעים בצורה של מערך בייטים. כשמוגדר קידוד הודעות JSON, מפתחות השורות מוחזרים כמחרוזות. כשמציינים את useBase64Rowkeys, מפתחות השורות מקודדים ב-Base64. אחרת, נעשה שימוש בערכת תווים שצוינה על ידי bigtableChangeStreamCharset כדי לפענח את הבייטים של מפתח השורה למחרוזת.
modType	סוג השינוי בשורה. משתמשים באחד מהערכים הבאים: SET_CELL, DELETE_CELLS או DELETE_FAMILY.
columnFamily	קבוצת העמודות שהושפעה משינוי השורה.
column	מגדיר העמודה שהושפע משינוי השורה. עבור סוג המוטציה DELETE_FAMILY, שדה העמודה לא מוגדר. הנתונים מגיעים בצורה של מערך בייטים. כשמוגדר קידוד הודעות JSON, העמודות מוחזרות כמחרוזות. אם מציינים את useBase64ColumnQualifier, שדה העמודה מקודד ב-Base64. אחרת, נעשה שימוש בערכת תווים שצוינה על ידי bigtableChangeStreamCharset כדי לפענח את הבייטים של מפתח השורה למחרוזת.
commitTimestamp	השעה שבה Bigtable מחיל את השינוי. הזמן נמדד במיקרו-שניות מאז ראשית זמן יוניקס (1 בינואר 1970 בשעה UTC).
timestamp	ערך חותמת הזמן של התא שהמוטציה משפיעה עליו. עבור סוגי המוטציות DELETE_CELLS ו-DELETE_FAMILY, חותמת הזמן לא מוגדרת. הזמן נמדד במיקרו-שניות מאז ראשית זמן יוניקס (1 בינואר 1970 בשעה UTC).
timestampFrom	מתאר את ההתחלה של מרווח הזמן של חותמות הזמן, כולל, לכל התאים שנמחקו על ידי המוטציה DELETE_CELLS. בסוגים אחרים של מוטציות, הערך של timestampFrom לא מוגדר. הזמן נמדד במיקרו-שניות מאז ראשית זמן יוניקס (1 בינואר 1970 בשעה UTC).
timestampTo	מתאר את הסוף הבלעדי של מרווח חותמות הזמן של כל התאים שנמחקו על ידי המוטציה DELETE_CELLS. בסוגים אחרים של מוטציות, הערך של timestampTo לא מוגדר.
isGC	ערך בוליאני שמציין אם המוטציה נוצרה על ידי מנגנון garbage collection של Bigtable.
tieBreaker	כששני שינויים נרשמים בו-זמנית על ידי אשכולות שונים של Bigtable, השינוי עם הערך הכי גבוה של tiebreaker מוחל על טבלת המקור. מוטציות עם ערכים נמוכים יותר של tiebreaker נפסלות.
value	הערך החדש שהוגדר על ידי השינוי. אם לא מגדירים את האפשרות stripValues של צינור העיבוד, הערך מוגדר לשינויים ב-SET_CELL. לסוגים אחרים של מוטציות, הערך לא מוגדר. הנתונים מגיעים בצורה של מערך בייטים. כשמוגדר קידוד של הודעות JSON, הערכים מוחזרים כמחרוזות. כשמציינים את useBase64Values, הערך מקודד בפורמט Base64. אחרת, נעשה שימוש בערכת תווים שצוינה על ידי bigtableChangeStreamCharset כדי לפענח את בייטים של הערך למחרוזת.
sourceInstance	השם של מופע Bigtable שרשם את השינוי. יכול להיות שזה יקרה כשמספר צינורות מעבירים שינויים ממופעים שונים לאותו נושא Pub/Sub.
sourceCluster	השם של אשכול Bigtable שרשם את השינוי. יכול להיות שיהיה צורך להשתמש בשיטה הזו כשכמה צינורות מעבירים שינויים בסטרימינג ממופעים שונים לאותו נושא Pub/Sub.
sourceTable	השם של טבלת Bigtable שקיבלה את השינוי. יכול להיות שיהיה צורך להשתמש בה אם כמה צינורות מעבירים שינויים בסטרימינג מטבלאות שונות לאותו נושא Pub/Sub.

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

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

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

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

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

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

• ‫messageEncoding: הקידוד של ההודעות שיפורסמו בנושא Pub/Sub. כשסכימת נושא היעד מוגדרת, קידוד ההודעה נקבע לפי הגדרות הנושא. הערכים הנתמכים הם: BINARY ו-JSON. ברירת המחדל היא JSON.
• ‫messageFormat: הקידוד של ההודעות לפרסום בנושא Pub/Sub. כשסכימת נושא היעד מוגדרת, קידוד ההודעה נקבע לפי הגדרות הנושא. הערכים הנתמכים: AVRO,‏ PROTOCOL_BUFFERS ו-JSON. ערך ברירת המחדל הוא JSON. כשמשתמשים בפורמט JSON, השדות rowKey,‏ column ו-value של ההודעה הם מחרוזות, והתוכן שלהם נקבע לפי אפשרויות צינור העיבוד useBase64Rowkeys,‏ useBase64ColumnQualifiers,‏ useBase64Values ו-bigtableChangeStreamCharset.
• ‫stripValues: אם הערך מוגדר כ-true, מוטציות SET_CELL מוחזרות בלי ערכים חדשים. ברירת המחדל היא false. הפרמטר הזה שימושי כשלא צריך שערך חדש יהיה נוכח, או כשערכים גדולים מאוד וחורגים ממגבלות הגודל של הודעות Pub/Sub.
• ‫dlqDirectory: הספרייה של תור ההודעות המתות. רשומות שלא עוברות עיבוד מאוחסנות בספרייה הזו. ברירת המחדל היא ספרייה מתחת למיקום הזמני של עבודת Dataflow. ברוב המקרים אפשר להשתמש בנתיב ברירת המחדל.
• ‫dlqRetryMinutes: מספר הדקות בין ניסיונות חוזרים של תור הודעות מתות. ברירת המחדל היא 10.
• ‫dlqMaxRetries: המספר המקסימלי של ניסיונות חוזרים של הודעות שהועברו לתור Dead Letter. ברירת המחדל היא 5.
• ‫useBase64Rowkeys: משמש עם קידוד הודעות JSON. אם הערך הוא true, השדה rowKey הוא מחרוזת בקידוד Base64. אחרת, rowKey נוצר באמצעות bigtableChangeStreamCharset כדי לפענח בייטים למחרוזת. ברירת המחדל היא false.
• ‫pubSubProjectId: מזהה הפרויקט ב-Bigtable. ברירת המחדל היא הפרויקט של משימת Dataflow.
• ‫useBase64ColumnQualifiers: משמש עם קידוד הודעות JSON. אם הערך הוא true, השדה column הוא מחרוזת בקידוד Base64. אחרת, העמודה נוצרת באמצעות bigtableChangeStreamCharset כדי לפענח בייטים למחרוזת. ברירת המחדל היא false.
• ‫useBase64Values: משמש עם קידוד הודעות JSON. אם הערך הוא true, שדה הערך הוא מחרוזת בקידוד Base64. אחרת, הערך נוצר באמצעות bigtableChangeStreamCharset כדי לפענח בייטים למחרוזת. ברירת המחדל היא false.
• ‫disableDlqRetries: מציין אם להשבית את הניסיונות החוזרים עבור תור ההודעות המתות. ברירת המחדל היא: false.
• ‫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 Pub/Sub 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_PubSub \
    --parameters \
bigtableReadInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableReadTableId=BIGTABLE_TABLE_ID,\
bigtableChangeStreamAppProfile=BIGTABLE_APPLICATION_PROFILE_ID,\
pubSubTopic=PUBSUB_TOPIC

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

• ‫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.
• ‫PUBSUB_TOPIC: שם נושא היעד ב-Pub/Sub

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_PubSub",
    "parameters": {
        "bigtableReadInstanceId": "BIGTABLE_INSTANCE_ID",
        "bigtableReadTableId": "BIGTABLE_TABLE_ID",
        "bigtableChangeStreamAppProfile": "BIGTABLE_APPLICATION_PROFILE_ID",
        "pubSubTopic": "PUBSUB_TOPIC"
    }
  }
}

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

• ‫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.
• ‫PUBSUB_TOPIC: שם נושא היעד ב-Pub/Sub

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

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.bigtablechangestreamstopubsub;

import com.google.cloud.Timestamp;
import com.google.cloud.bigtable.data.v2.models.ChangeStreamMutation;
import com.google.cloud.bigtable.data.v2.models.ChangeStreamMutation.MutationType;
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.pubsub.v1.SchemaServiceClient;
import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions;
import com.google.cloud.teleport.v2.bigtable.utils.UnsupportedEntryException;
import com.google.cloud.teleport.v2.cdc.dlq.DeadLetterQueueManager;
import com.google.cloud.teleport.v2.cdc.dlq.StringDeadLetterQueueSanitizer;
import com.google.cloud.teleport.v2.coders.FailsafeElementCoder;
import com.google.cloud.teleport.v2.options.BigtableChangeStreamsToPubSubOptions;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.FailsafePublisher.PublishModJsonToTopic;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.MessageEncoding;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.MessageFormat;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.Mod;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.ModType;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.PubSubDestination;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.TestChangeStreamMutation;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.schemautils.PubSubUtils;
import com.google.cloud.teleport.v2.transforms.DLQWriteTransform;
import com.google.cloud.teleport.v2.utils.BigtableSource;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.Encoding;
import com.google.pubsub.v1.GetTopicRequest;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.Schema;
import com.google.pubsub.v1.SchemaName;
import com.google.pubsub.v1.Topic;
import com.google.pubsub.v1.TopicName;
import com.google.pubsub.v1.ValidateMessageRequest;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Base64;
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.coders.CoderRegistry;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Flatten;
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.beam.sdk.values.PCollectionList;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.Duration;
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 PubsubMessage and
 * inserted into Pub/Sub topic.
 */
@Template(
    name = "Bigtable_Change_Streams_to_PubSub",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Bigtable Change Streams to PubSub",
    description =
        "Streaming pipeline. Streams Bigtable data change records and writes them into PubSub using Dataflow Runner V2.",
    optionsClass = BigtableChangeStreamsToPubSubOptions.class,
    optionsOrder = {
      BigtableChangeStreamsToPubSubOptions.class,
      BigtableCommonOptions.ReadChangeStreamOptions.class,
      BigtableCommonOptions.ReadOptions.class
    },
    skipOptions = {
      "bigtableReadAppProfile",
      "bigtableAdditionalRetryCodes",
      "bigtableRpcAttemptTimeoutMs",
      "bigtableRpcTimeoutMs"
    },
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-bigtable-change-streams-to-pubsub",
    flexContainerName = "googlecloud-to-googlecloud",
    contactInformation = "https://cloud.google.com/support",
    streaming = true,
    supportsAtLeastOnce = true)
public final class BigtableChangeStreamsToPubSub {

  /** String/String Coder for {@link FailsafeElement}. */
  public static final FailsafeElementCoder<String, String> FAILSAFE_ELEMENT_CODER =
      FailsafeElementCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of());

  private static final Logger LOG = LoggerFactory.getLogger(BigtableChangeStreamsToPubSub.class);

  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

  private static final TupleTag<FailsafeElement<String, String>> INVALID_MODS_TAG =
      new TupleTag<FailsafeElement<String, String>>("invalidMods") {};
  private static final TupleTag<FailsafeElement<String, String>> VALID_MODS_TAG =
      new TupleTag<FailsafeElement<String, String>>("validMods") {};

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

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

    run(options);
  }

  private static void validateOptions(BigtableChangeStreamsToPubSubOptions options) {
    if (options.getDlqRetryMinutes() <= 0) {
      throw new IllegalArgumentException("dlqRetryMinutes must be positive.");
    }
    if (options.getDlqMaxRetries() < 0) {
      throw new IllegalArgumentException("dlqMaxRetries cannot be negative.");
    }
    if (options.getBigtableReadChangeStreamTimeoutMs() != null
        && options.getBigtableReadChangeStreamTimeoutMs() <= 0) {
      throw new IllegalArgumentException(
          "bigtableReadChangeStreamTimeoutMs must be greater than 0.");
    }
  }

  private static void setOptions(BigtableChangeStreamsToPubSubOptions 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(BigtableChangeStreamsToPubSubOptions options) {
    setOptions(options);
    validateOptions(options);

    String bigtableProject = getBigtableProjectId(options);

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

    Topic topic = null;
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
      GetTopicRequest request =
          GetTopicRequest.newBuilder()
              .setTopic(
                  TopicName.ofProjectTopicName(
                          getPubSubProjectId(options), options.getPubSubTopic())
                      .toString())
              .build();
      topic = topicAdminClient.getTopic(request);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }

    try {
      if (!validateSchema(topic, options, sourceInfo)) {
        final String errorMessage = "Configured topic doesn't accept messages of configured format";
        throw new IllegalArgumentException(errorMessage);
      }

    } catch (Exception e) {
      throw new IllegalArgumentException(e);
    }

    PubSubDestination destinationInfo = newPubSubDestination(options, topic);
    PubSubUtils pubSub = new PubSubUtils(sourceInfo, destinationInfo);

    /*
     * Stages:
     * 1) Read {@link ChangeStreamMutation} from change stream.
     * 2) Create {@link FailsafeElement} of {@link Mod} JSON and merge from:
     *    - {@link ChangeStreamMutation}.
     *    - GCS Dead letter queue.
     * 3) Convert {@link Mod} JSON into PubsubMessage and publish it to PubSub.
     * 4) Write Failures from 2) and 3) to GCS dead letter queue.
     */
    // Step 1
    Pipeline pipeline = Pipeline.create(options);

    // Register the coders for pipeline
    CoderRegistry coderRegistry = pipeline.getCoderRegistry();
    coderRegistry.registerCoderForType(
        FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor(), FAILSAFE_ELEMENT_CODER);

    DeadLetterQueueManager dlqManager = buildDlqManager(options);

    String dlqDirectory = dlqManager.getRetryDlqDirectoryWithDateTime();
    String tempDlqDirectory = dlqManager.getRetryDlqDirectory() + "tmp/";

    if (options.getDisableDlqRetries()) {
      LOG.info(
          "Disabling retries for the DLQ, directly writing into severe DLQ: {}",
          dlqManager.getSevereDlqDirectoryWithDateTime());
      dlqDirectory = dlqManager.getSevereDlqDirectoryWithDateTime();
      tempDlqDirectory = dlqManager.getSevereDlqDirectory() + "tmp/";
    }

    BigtableIO.ReadChangeStream readChangeStream =
        BigtableIO.readChangeStream()
            .withChangeStreamName(options.getBigtableChangeStreamName())
            .withExistingPipelineOptions(
                options.getBigtableChangeStreamResume()
                    ? BigtableIO.ExistingPipelineOptions.RESUME_OR_FAIL
                    : BigtableIO.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());
    }

    if (options.getBigtableReadChangeStreamTimeoutMs() != null) {
      readChangeStream =
          readChangeStream.withReadChangeStreamTimeout(
              Duration.millis(options.getBigtableReadChangeStreamTimeoutMs()));
    }

    // Step 2: just return the output for sending to pubSub and DLQ
    PCollection<ChangeStreamMutation> dataChangeRecord =
        pipeline
            .apply("Read from Cloud Bigtable Change Streams", readChangeStream)
            .apply(Values.create());

    PCollection<FailsafeElement<String, String>> sourceFailsafeModJson =
        dataChangeRecord
            .apply(
                "ChangeStreamMutation To Mod JSON",
                ParDo.of(new ChangeStreamMutationToModJsonFn(sourceInfo)))
            .apply(
                "Wrap Mod JSON In FailsafeElement",
                ParDo.of(
                    new DoFn<String, FailsafeElement<String, String>>() {
                      @ProcessElement
                      public void process(
                          @Element String input,
                          OutputReceiver<FailsafeElement<String, String>> receiver) {
                        receiver.output(FailsafeElement.of(input, input));
                      }
                    }))
            .setCoder(FAILSAFE_ELEMENT_CODER);

    PCollectionTuple dlqModJson =
        dlqManager.getReconsumerDataTransform(
            pipeline.apply(dlqManager.dlqReconsumer(options.getDlqRetryMinutes())));

    PCollection<FailsafeElement<String, String>> retryableDlqFailsafeModJson = null;
    if (options.getDisableDlqRetries()) {
      retryableDlqFailsafeModJson = pipeline.apply(Create.empty(FAILSAFE_ELEMENT_CODER));
    } else {
      retryableDlqFailsafeModJson =
          dlqModJson.get(DeadLetterQueueManager.RETRYABLE_ERRORS).setCoder(FAILSAFE_ELEMENT_CODER);
    }

    PCollection<FailsafeElement<String, String>> failsafeModJson =
        PCollectionList.of(sourceFailsafeModJson)
            .and(retryableDlqFailsafeModJson)
            .apply("Merge Source And DLQ Mod JSON", Flatten.pCollections());

    PublishModJsonToTopic publishModJsonToTopic =
        new PublishModJsonToTopic(pubSub, VALID_MODS_TAG, INVALID_MODS_TAG);

    PCollectionTuple failedToPublish =
        failsafeModJson.apply("Publish Mod JSON To Pubsub", publishModJsonToTopic);

    PCollection<String> transformDlqJson =
        failedToPublish
            .get(VALID_MODS_TAG)
            .setCoder(FAILSAFE_ELEMENT_CODER)
            .apply(
                "Failed Mod JSON During Table Row Transformation",
                MapElements.via(new StringDeadLetterQueueSanitizer()));

    PCollectionList.of(transformDlqJson)
        .apply("Merge Failed Mod JSON From Transform And PubSub", Flatten.pCollections())
        .apply(
            "Write Failed Mod JSON To DLQ",
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqDirectory)
                .withTmpDirectory(tempDlqDirectory)
                .setIncludePaneInfo(true)
                .build());

    PCollection<FailsafeElement<String, String>> nonRetryableDlqModJsonFailsafe =
        PCollectionList.of(
                dlqModJson
                    .get(DeadLetterQueueManager.PERMANENT_ERRORS)
                    .setCoder(FAILSAFE_ELEMENT_CODER))
            .and(failedToPublish.get(INVALID_MODS_TAG).setCoder(FAILSAFE_ELEMENT_CODER))
            .apply("Merge Reconsume And Invalid Mods", Flatten.pCollections());
    LOG.info(
        "DLQ manager severe DLQ directory with date time: {}",
        dlqManager.getSevereDlqDirectoryWithDateTime());
    LOG.info("DLQ manager severe DLQ directory: {}", dlqManager.getSevereDlqDirectory() + "tmp/");
    nonRetryableDlqModJsonFailsafe
        .apply(
            "Write Mod JSON With Non-retriable Error To DLQ",
            MapElements.via(new StringDeadLetterQueueSanitizer()))
        .setCoder(StringUtf8Coder.of())
        .apply(
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqManager.getSevereDlqDirectoryWithDateTime())
                .withTmpDirectory(dlqManager.getSevereDlqDirectory() + "tmp/")
                .setIncludePaneInfo(true)
                .build());

    return pipeline.run();
  }

  private static PubSubDestination newPubSubDestination(
      BigtableChangeStreamsToPubSubOptions options, Topic topic) {
    return new PubSubDestination(
        getPubSubProjectId(options),
        options.getPubSubTopic(),
        topic,
        options.getMessageFormat(),
        options.getMessageEncoding(),
        options.getUseBase64Rowkeys(),
        options.getUseBase64ColumnQualifiers(),
        options.getUseBase64Values(),
        options.getStripValues());
  }

  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(
      BigtableChangeStreamsToPubSubOptions 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("DLQ directory: {}", dlqDirectory);
    return DeadLetterQueueManager.create(dlqDirectory, options.getDlqMaxRetries());
  }

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

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

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

  private static Boolean validateSchema(
      Topic topic, BigtableChangeStreamsToPubSubOptions options, BigtableSource source)
      throws Exception {
    String messageFormatPath = topic.getSchemaSettings().getSchema();
    if (topic.getSchemaSettings().getSchema().isEmpty()) {
      validateIncompatibleEncoding(options);
      LOG.info(
          "Topic has no schema configured, pipeline will use message format: {}, message encoding: {}",
          options.getMessageFormat(),
          options.getMessageEncoding());
      return true;
    } else {
      SchemaName schemaName = SchemaName.parse(topic.getSchemaSettings().getSchema());
      Schema schema;
      try (SchemaServiceClient schemaServiceClient = SchemaServiceClient.create()) {
        schema = schemaServiceClient.getSchema(schemaName);
      }

      options.setMessageEncoding(toMessageEncoding(topic.getSchemaSettings().getEncoding()));

      Schema.Type schemaType = schema.getType();
      switch (schemaType) {
        case AVRO:
          options.setMessageFormat(MessageFormat.AVRO);
          validateNoUseOfBase64(options);
          break;
        case PROTOCOL_BUFFER:
          if (options.getMessageEncoding() == MessageEncoding.JSON) {
            options.setMessageFormat(MessageFormat.JSON);
          } else {
            options.setMessageFormat(MessageFormat.PROTOCOL_BUFFERS);
            validateNoUseOfBase64(options);
          }
          break;
        case TYPE_UNSPECIFIED:
        case UNRECOGNIZED:
          // Not overriding messageFormat, will try what customer configured or the default if
          // not configured
          break;
        default:
          throw new IllegalArgumentException("Topic schema type is not supported: " + schemaType);
      }

      LOG.info("Topic has schema configured: {}", topic.getSchemaSettings().getSchema());
      LOG.info(
          "Pipeline will use message format: {}, message encoding: {}",
          options.getMessageFormat(),
          options.getMessageEncoding());

      PubSubDestination destination = newPubSubDestination(options, topic);
      PubSubUtils pubSub = new PubSubUtils(source, destination);

      ByteString testChangeMessageData = createTestChangeMessage(pubSub).getData();
      Encoding encoding = toPubSubEncoding(options.getMessageEncoding());
      try (SchemaServiceClient schemaServiceClient = SchemaServiceClient.create()) {
        String testMessageEncoded = toBase64String(testChangeMessageData);
        LOG.info("Validating a test message (Base64 encoded): {}", testMessageEncoded);
        ValidateMessageRequest request =
            ValidateMessageRequest.newBuilder()
                .setParent("projects/" + pubSub.getDestination().getPubSubProject())
                .setEncoding(encoding)
                .setMessage(testChangeMessageData)
                .setName(messageFormatPath)
                .build();
        schemaServiceClient.validateMessage(request);
        LOG.info("Test message successfully validated.");
      } catch (Exception e) {
        throw new IllegalArgumentException("Failed to validate test message", e);
      }
    }
    return true;
  }

  private static void validateNoUseOfBase64(BigtableChangeStreamsToPubSubOptions options) {
    if (options.getUseBase64Values()) {
      throw new IllegalArgumentException(
          "useBase64Values values can only be used with topics accepting JSON messages");
    }
    if (options.getUseBase64Rowkeys()) {
      throw new IllegalArgumentException(
          "useBase64Rowkeys values can only be used with topics accepting JSON messages");
    }
    if (options.getUseBase64ColumnQualifiers()) {
      throw new IllegalArgumentException(
          "useBase64ColumnQualifiers values can only be used with topics accepting JSON messages");
    }
  }

  private static void validateIncompatibleEncoding(BigtableChangeStreamsToPubSubOptions options) {
    if (options.getMessageEncoding() == MessageEncoding.BINARY
        && options.getMessageFormat() == MessageFormat.JSON) {
      throw new IllegalArgumentException(
          "JSON message format is incompatible with BINARY message encoding");
    }
  }

  private static MessageEncoding toMessageEncoding(Encoding encoding) {
    if (encoding == null) {
      return MessageEncoding.JSON;
    }
    switch (encoding) {
      case JSON:
      case ENCODING_UNSPECIFIED:
      case UNRECOGNIZED:
        return MessageEncoding.JSON;
      case BINARY:
        return MessageEncoding.BINARY;
      default:
        throw new IllegalArgumentException("Topic has unsupported message encoding: " + encoding);
    }
  }

  private static String toBase64String(ByteString testChangeMessageData) {
    return Base64.getEncoder().encodeToString(testChangeMessageData.toByteArray());
  }

  private static Encoding toPubSubEncoding(MessageEncoding messageEncoding) {
    switch (messageEncoding) {
      case BINARY:
        return Encoding.BINARY;
      case JSON:
        return Encoding.JSON;
      default:
        throw new IllegalArgumentException("Unexpected message encoding: " + messageEncoding);
    }
  }

  private static PubsubMessage createTestChangeMessage(PubSubUtils pubSub) throws Exception {
    SetCell setCell =
        SetCell.create(
            "test_column_family",
            ByteString.copyFrom("test_column", Charset.defaultCharset()),
            1000L, // timestamp
            ByteString.copyFrom("test_value", Charset.defaultCharset()));

    TestChangeStreamMutation mutation =
        new TestChangeStreamMutation(
            "test_rowkey",
            MutationType.USER,
            "source_cluster",
            java.time.Instant.now(), // commit timestamp
            1, // tiebreaker
            "token",
            java.time.Instant.now(), // low watermark
            setCell);

    Mod mod = new Mod(pubSub.getSource(), mutation, setCell);

    switch (pubSub.getDestination().getMessageFormat()) {
      case AVRO:
        return pubSub.mapChangeJsonStringToPubSubMessageAsAvro(mod.getChangeJson());
      case PROTOCOL_BUFFERS:
        return pubSub.mapChangeJsonStringToPubSubMessageAsProto(mod.getChangeJson());
      case JSON:
        return pubSub.mapChangeJsonStringToPubSubMessageAsJson(mod.getChangeJson());
      default:
        throw new IllegalArgumentException(
            "Unexpected message format: " + pubSub.getDestination().getMessageFormat());
    }
  }

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

    ChangeStreamMutationToModJsonFn(BigtableSource source) {
      this.sourceInfo = source;
    }

    private static String toJsonString(Mod mod, ChangeStreamMutation inputMutation) {
      try {
        return mod.toJson();
      } catch (IOException e) {
        // Ignore exception and print bad format.
        return String.format("\"%s\"", inputMutation);
      }
    }

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

        switch (modType) {
          case SET_CELL:
            SetCell setCell = (SetCell) entry;
            if (!sourceInfo.isIgnoredColumnFamily(setCell.getFamilyName())
                && !sourceInfo.isIgnoredColumn(
                    setCell.getFamilyName(),
                    setCell.getQualifier().toString(Charset.forName(sourceInfo.getCharset())))) {
              Mod mod = new Mod(sourceInfo, input, setCell);
              receiver.output(toJsonString(mod, input));
            }
            break;
          case DELETE_CELLS:
            DeleteCells deleteCells = (DeleteCells) entry;
            if (!sourceInfo.isIgnoredColumnFamily(deleteCells.getFamilyName())
                && !sourceInfo.isIgnoredColumn(
                    deleteCells.getFamilyName(),
                    deleteCells
                        .getQualifier()
                        .toString(Charset.forName(sourceInfo.getCharset())))) {
              Mod mod = new Mod(sourceInfo, input, deleteCells);
              receiver.output(toJsonString(mod, input));
            }
            break;
          case DELETE_FAMILY:
            DeleteFamily deleteFamily = (DeleteFamily) entry;
            if (!sourceInfo.isIgnoredColumnFamily(deleteFamily.getFamilyName())) {
              Mod mod = new Mod(sourceInfo, input, deleteFamily);
              receiver.output(toJsonString(mod, input));
            }
            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 DLQ directory. "
                    + "Please update your Dataflow template with the latest template version");
        }
      }
    }

    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 סיפקה