תבנית של סנכרון שינויים בזרמי נתונים מ-Spanner ל-BigQuery

תבנית סנכרון שינויים בזרמי נתונים מ-Spanner ל-BigQuery היא צינור להזרמת נתונים שמזרים רשומות של שינויים בנתונים מ-Spanner וכותב אותן לטבלאות ב-BigQuery באמצעות Dataflow Runner V2.

הערה: מחבר Spanner סנכרון שינויים בזרמי נתונים לא תומך בהפסקת העבודה של משימה. כדי להפסיק את הצינור, מבטלים את העבודה במקום לנקז אותה. הפסקת העבודה של משימה באמצעות מחבר של Spanner לסנכרון שינויים בזרמי נתונים יכולה לגרום לכך שהמשימה תיתקע במצב של הפסקת עבודה. מידע נוסף זמין במאמר בנושא הפסקת פעילות במסמכי התיעוד של Spanner.

עמודות שלא נצפו לא נכללות בשורה ב-BigQuery. כל השינויים ב-Spanner שקטנים מסימן המים של Dataflow מוחלים בהצלחה על הטבלאות ב-BigQuery או מאוחסנים בתור להודעות שלא ניתן למסור (dead-letter queue) לניסיון חוזר. השורה ב-BigQuery מוכנסת לא לפי הסדר, בהשוואה לסדר של חותמת הזמן המקורית של ביצוע השינויים ב-Spanner.

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

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

• ‫_metadata_spanner_mod_type: סוג השינוי (הוספה, עדכון או מחיקה) של טרנזקציית Spanner. המידע חולץ מרשומה של שינוי נתונים בזרם השינויים.
• ‫_metadata_spanner_table_name: שם הטבלה ב-Spanner. השדה הזה לא מייצג את השם של טבלת המטא-נתונים של המחבר.
• ‫_metadata_spanner_commit_timestamp: חותמת הזמן של ביצוע השינוי ב-Spanner, כלומר השעה שבה השינוי בוצע. הערך הזה מחולץ מרשומה של שינוי נתונים בזרם השינויים.
• ‫_metadata_spanner_server_transaction_id: מחרוזת ייחודית גלובלית שמייצגת את הטרנזקציה ב-Spanner שבה השינוי בוצע. אפשר להשתמש בערך הזה רק בהקשר של עיבוד רשומות בזרם שינויים. הוא לא קשור למזהה העסקה ב-API של Spanner. הערך הזה מחולץ מרשומה של שינוי נתונים בזרם השינויים.
• ‫_metadata_spanner_record_sequence: המספר הסידורי של הרשומה בתוך טרנזקציית Spanner. מספרי הרצף מובטחים להיות ייחודיים ולעלות באופן מונוטוני, אבל לא בהכרח רציפים, במסגרת עסקה. הערך הזה מחולץ מרשומה של שינוי נתונים בזרם השינויים.
• ‫_metadata_spanner_is_last_record_in_transaction_in_partition: מציין אם הרשומה היא הרשומה האחרונה של טרנזקציית Spanner במחיצה הנוכחית. הערך הזה מחולץ מרשומה של שינוי נתונים בזרם השינויים.
• ‫_metadata_spanner_number_of_records_in_transaction: מספר הרשומות של שינוי הנתונים שכלולות בעסקת Spanner בכל המחיצות של עדכון השינויים. הערך הזה מחולץ מרשומה של שינוי נתונים בזרם השינויים.
• ‫_metadata_spanner_number_of_partitions_in_transaction: מספר המחיצות שמחזירות רשומות של שינויים בנתונים של טרנזקציית Spanner. הערך הזה מחולץ מרשומה של שינוי נתונים בזרם השינויים.
• ‫_metadata_big_query_commit_timestamp: חותמת הזמן של השליחה (commit) כששורה מוכנסת ל-BigQuery. אם הערך של useStorageWriteApi הוא true, העמודה הזו לא נוצרת באופן אוטומטי בטבלת יומן השינויים על ידי צינור הנתונים. במקרה כזה, צריך להוסיף את העמודה הזו באופן ידני לטבלת יומן השינויים ולהגדיר את CURRENT_TIMESTAMP כערך ברירת המחדל שלה, אם צריך.

כשמשתמשים בתבנית הזו, חשוב לשים לב לפרטים הבאים:

• אפשר להשתמש בתבנית הזו כדי להפיץ עמודות חדשות בטבלאות קיימות או בטבלאות חדשות מ-Spanner ל-BigQuery. מידע נוסף זמין במאמר בנושא הוספה של טבלאות או עמודות למעקב.
• בסוגי הלכידה של הערכים OLD_AND_NEW_VALUES ו-NEW_VALUES, כשרשומה של שינוי נתונים מכילה שינוי מסוג UPDATE, התבנית צריכה לבצע קריאה של נתונים לא עדכניים מ-Spanner בחותמת הזמן של ביצוע השינוי ברשומה, כדי לאחזר את העמודות שלא השתנו אבל נמצאות במעקב. חשוב להגדיר את 'version_retention_period' (תקופת השמירה של הגרסה) במסד הנתונים בצורה נכונה כדי לאפשר קריאה בעבר. במקרה של סוג לכידת הערך NEW_ROW, התבנית יעילה יותר כי רשומת שינוי הנתונים לוכדת את השורה החדשה המלאה, כולל עמודות שלא עודכנו בבקשות UPDATE, והתבנית לא צריכה לבצע קריאה של נתונים לא עדכניים.
• כדי לצמצם את זמן האחזור ברשת ואת עלויות התעבורה ברשת, מריצים את משימת Dataflow מאותו אזור שבו נמצאים מופע Spanner או טבלאות BigQuery. אם אתם משתמשים במקורות, ביעדים, במיקומים של קבצים זמניים או במיקומים של קבצים להעברה שנמצאים מחוץ לאזור של העבודה, יכול להיות שהנתונים שלכם יישלחו בין אזורים. מידע נוסף זמין במאמר בנושא אזורי Dataflow.
• התבנית הזו תומכת בכל סוגי הנתונים התקינים ב-Spanner. אם הסוג ב-BigQuery מדויק יותר מהסוג ב-Spanner, יכול להיות שיהיה אובדן דיוק במהלך ההמרה. באופן ספציפי:
  • בסוג JSON של Spanner, הסדר של חברי אובייקט הוא סדר לקסיקוגרפי, אבל אין הבטחה כזו לגבי סוג JSON של BigQuery.
  • ‫Spanner תומך בסוג חותמת זמן של ננו-שניות, אבל BigQuery תומך רק בסוג חותמת זמן של מיקרו-שניות.

מידע נוסף על סנכרון שינויים בזרמי נתונים, איך יוצרים צינורות עיבוד נתונים של Dataflow לסנכרון שינויים בזרמי נתונים ושיטות מומלצות

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

• מופע Spanner חייב להתקיים לפני שמריצים את צינור הנתונים.
• מסד הנתונים של Spanner צריך להתקיים לפני שמריצים את צינור הנתונים.
• מופע המטא-נתונים של Spanner צריך להתקיים לפני שמריצים את צינור העיבוד.
• מסד הנתונים של המטא-נתונים של Spanner צריך להתקיים לפני שמריצים את צינור העיבוד.
• סנכרון שינויים בזרמי נתונים של Spanner חייב להתקיים לפני שמריצים את צינור הנתונים.
• המערכת תומכת רק בסנכרון שינויים בזרמי נתונים שצופים בטבלאות תחת הסכימה שמוגדרת כברירת מחדל.
• סנכרון שינויים בזרמי נתונים שמתבצעים בטבלאות שנמצאות בסכימות אחרות גורמים ל[שגיאה שלא נמצאה טבלה](https://github.com/GoogleCloudPlatform/DataflowTemplates/issues/2622).
• מערך הנתונים ב-BigQuery צריך להתקיים לפני שמריצים את צינור הנתונים.

טיפול בהוספה של טבלאות או עמודות למעקב

בקטע הזה מתוארות שיטות מומלצות לטיפול בהוספה של טבלאות ועמודות למעקב ב-Spanner בזמן שהצינור פועל. הגרסה הכי ישנה של תבנית שנתמכת בתכונה הזו היא 2024-09-19-00_RC00.

• לפני שמוסיפים עמודה חדשה להיקף של זרם שינויים ב-Spanner, קודם מוסיפים את העמודה לטבלת יומן השינויים ב-BigQuery. העמודה שמוסיפים צריכה להיות מאותו סוג נתונים ולהיות NULLABLE. צריך להמתין לפחות 10 דקות לפני שממשיכים ליצור את העמודה או הטבלה החדשות ב-Spanner. אם תכתבו לעמודה החדשה בלי לחכות, יכול להיות שרשומה לא מעובדת עם קוד שגיאה לא תקין תתווסף לתור ההודעות שלא הועברו.

  בסוגי לכידת ערכים NEW_ROW או NEW_ROW_AND_OLD_VALUES, אפשר להוסיף עמודה חדשה עם ערך ברירת מפל במהלך צינור פעיל. כדי לעשות זאת, קודם מוסיפים את העמודה לטבלה ב-BigQuery ואז מוסיפים אותה לטבלת Spanner.

  במקרה של NEW_VALUES או OLD_AND_NEW_VALUES, הוספה של עמודה חדשה עם ערך ברירת מחדל לצינור פעיל עלולה לגרום לאובדן נתונים. כדי להימנע מאובדן נתונים, צריך לייצא את הנתונים באופן ידני ולייבא אותם באמצעות התהליך הבא:

1. מפסיקים את צינור עיבוד הנתונים של Dataflow.
2. מוסיפים את העמודה לטבלה ב-BigQuery.
3. מוסיפים את העמודה עם ערך ברירת מחדל ב-Spanner.
4. ממתינים לסיום השינוי בסכימה של Spanner ולמילוי החסר.
5. מייצאים את נתוני הטבלה של Spanner בחותמת זמן אחרי שמילוי החוסרים מסתיים.
6. מייבאים את הנתונים לטבלת BigQuery, ומוודאים שהטיפול בעמודה החדשה מתבצע בצורה נכונה.
7. מפעילים את צינור הנתונים של Dataflow מאותה חותמת זמן שבה השתמשתם לייצוא.
הערה: כשמוסיפים עמודה חדשה עם ערך ברירת מחדל לטבלת Spanner קיימת, מערכת Spanner מבצעת פעולת מילוי חוזר כדי לאכלס את כל השורות הקיימות בערך ברירת המחדל הזה. הפעולה הזו יכולה לקחת זמן, בהתאם לגודל הטבלה. אתם יכולים לעקוב אחרי ההתקדמות של שינוי הסכימה, כולל מילוי החוסרים, על ידי שליחת שאילתה לתצוגה information_schema.COLUMNS ב-Spanner. ההשלמה תסתיים כשהערך SPANNER_STATE בעמודה החדשה יהיה COMMITTED.
• כדי להוסיף טבלה חדשה, קודם מוסיפים את הטבלה במסד הנתונים של Spanner. הטבלה נוצרת באופן אוטומטי ב-BigQuery כשהצינור מקבל רשומה לטבלה החדשה.
• אחרי שמוסיפים את העמודות או הטבלאות החדשות במסד הנתונים של Spanner, צריך לשנות את זרם השינויים כדי לעקוב אחרי העמודות או הטבלאות החדשות שרוצים, אם המערכת לא עוקבת אחריהן באופן מרומז.
• התבנית לא מוחקת טבלאות או עמודות מ-BigQuery. אם עמודה נמחקת מטבלת Spanner, ערכי null מאוכלסים בעמודות של יומן השינויים ב-BigQuery עבור רשומות שנוצרו אחרי שהעמודות נמחקו מטבלת Spanner, אלא אם מוחקים את העמודה מ-BigQuery באופן ידני.
• התבנית לא תומכת בעדכונים של סוג העמודה. למרות שב-Spanner אפשר לשנות עמודה מסוג STRING לעמודה מסוג BYTES או עמודה מסוג BYTES לעמודה מסוג STRING, אי אפשר לשנות את סוג הנתונים של עמודה קיימת או להשתמש באותו שם עמודה עם סוגי נתונים שונים ב-BigQuery. אם משמיטים עמודה ויוצרים אותה מחדש עם אותו שם אבל סוג אחר ב-Spanner, יכול להיות שהנתונים ייכתבו בעמודה הקיימת ב-BigQuery, אבל הסוג לא ישתנה.
• התבנית הזו לא תומכת בעדכונים של מצב העמודה. עמודות המטא-נתונים שמשוכפלות ל-BigQuery מוגדרות למצב REQUIRED. כל שאר העמודות שמשוכפלות ל-BigQuery מוגדרות כ-NULLABLE, בלי קשר להגדרה שלהן כ-NULLABLE בטבלת Spanner.NOT NULL אי אפשר לעדכן את העמודות NULLABLE למצב REQUIRED ב-BigQuery.
• אי אפשר לשנות את סוג לכידת הערך של עדכון נתונים בזמן שהצינורות פועלים.

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

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

• ‫spannerInstanceId: מכונת Spanner שממנה יתבצע קריאה של סנכרון שינויים בזרמי נתונים.
• ‫spannerDatabase: מסד הנתונים של Spanner שממנו ייקראו סנכרון שינויים בזרמי נתונים.
• ‫spannerMetadataInstanceId: מופע Spanner לשימוש בטבלת המטא-נתונים של מחבר הנתונים של סנכרון שינויים בזרמי נתונים.
• ‫spannerMetadataDatabase: מסד הנתונים של Spanner שבו יש להשתמש עבור טבלת המטא-נתונים של מחבר סנכרון שינויים בזרמי נתונים.
• ‫spannerChangeStreamName: השם של סנכרון שינויים בזרמי נתונים ב-Spanner שממנו רוצים לקרוא.
• ‫bigQueryDataset: מערך הנתונים של BigQuery לפלט של סנכרון שינויים בזרמי נתונים.

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

• ‫spannerProjectId: הפרויקט שממנו יקראו את נתוני סנכרון שינויים בזרמי נתונים. הערך הזה הוא גם הפרויקט שבו נוצרת טבלת המטא-נתונים של מחבר סנכרון שינויים בזרמי נתונים. ערך ברירת המחדל של הפרמטר הזה הוא הפרויקט שבו צינור ה-Dataflow פועל.
• ‫spannerDatabaseRole: תפקיד מסד הנתונים ב-Spanner שבו יש להשתמש כשמריצים את התבנית. הפרמטר הזה נדרש רק אם משתמש ה-IAM שמריץ את התבנית הוא משתמש עם בקרת גישה ברמת דיוק גבוהה. לתפקיד במסד הנתונים צריכה להיות הרשאת SELECT בסנכרון שינויים בזרמי נתונים והרשאת EXECUTE בפונקציית הקריאה של סנכרון שינויים בזרמי נתונים. מידע נוסף זמין במאמר בנושא בקרת גישה ברמת גרנולריות גבוהה לסנכרון שינויים בזרמי נתונים (https://cloud.google.com/spanner/docs/fgac-change-streams).
• ‫spannerMetadataTableName: שם טבלת המטא-נתונים של מחבר Spanner לשימוש בסנכרון שינויים בזרמי נתונים. אם לא מספקים את המידע הזה, נוצרת באופן אוטומטי טבלת מטא-נתונים של מחבר Spanner לסנכרון שינויים בזרמי נתונים במהלך זרימת הצינור. כשמעדכנים צינור קיים, צריך לציין את הפרמטר הזה. אחרת, אל תציינו את הפרמטר הזה.
• ‫rpcPriority: עדיפות הבקשה לשיחות Spanner. הערך חייב להיות אחד מהערכים הבאים: HIGH, MEDIUM או LOW. ערך ברירת המחדל הוא HIGH.
• ‫spannerHost: נקודת הקצה של Cloud Spanner שאליה מתבצעת קריאה בתבנית. היא משמשת לבדיקה בלבד. לדוגמה, https://batch-spanner.googleapis.com.
• ‫startTimestamp: תאריך ושעת ההתחלה (https://datatracker.ietf.org/doc/html/rfc3339), כולל, לשימוש בקריאת סנכרון שינויים בזרמי נתונים. Ex-2021-10-12T07:20:50.52Z. ברירת המחדל היא חותמת הזמן של תחילת הצינור, כלומר השעה הנוכחית.
• ‫endTimestamp: תאריך ושעה לסיום (https://datatracker.ietf.org/doc/html/rfc3339), כולל, לשימוש בקריאת נתוני שינויים.דוגמה: ‎2021-10-12T07:20:50.52Z. ברירת המחדל היא זמן אינסופי בעתיד.
• ‫bigQueryProjectId: הפרויקט ב-BigQuery. ערך ברירת המחדל הוא הפרויקט של משימת Dataflow.
• ‫bigQueryChangelogTableNameTemplate: התבנית של השם של טבלת BigQuery שמכילה את יומן השינויים. ברירת המחדל היא: ‎{_metadata_spanner_table_name}_changelog.
• ‫deadLetterQueueDirectory: הנתיב לשמירת רשומות שלא עברו עיבוד. נתיב ברירת המחדל הוא ספרייה במיקום הזמני של משימת Dataflow. בדרך כלל ערך ברירת המחדל מספיק.
• ‫dlqRetryMinutes: מספר הדקות בין ניסיונות חוזרים של תור הודעות מתות. ערך ברירת המחדל הוא 10.
• ‫ignoreFields: רשימה מופרדת בפסיקים של שדות (תלוי אותיות רישיות) להתעלמות. יכול להיות שהשדות האלה הם שדות של טבלאות שנצפו, או שדות מטא-נתונים שנוספו על ידי צינור הנתונים. שדות שהמערכת מתעלמת מהם לא מוכנסים ל-BigQuery. כשמתעלמים מהשדה ‎_metadata_spanner_table_name, מתעלמים גם מהפרמטר bigQueryChangelogTableNameTemplate. ברירת המחדל היא ריק.
• ‫disableDlqRetries: מציין אם להשבית את הניסיונות החוזרים עבור תור ההודעות המתות. ברירת המחדל היא: false.
• ‫useStorageWriteApi: אם הערך הוא true, צינור הנתונים משתמש ב-BigQuery Storage Write API‏ (https://cloud.google.com/bigquery/docs/write-api). ערך ברירת המחדל הוא false. מידע נוסף זמין במאמר בנושא שימוש ב-Storage Write API‏ (https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api).
• ‫useStorageWriteApiAtLeastOnce: כשמשתמשים ב-Storage Write API, המאפיין הזה מציין את סמנטיקת הכתיבה. כדי להשתמש בסמנטיקה של 'לפחות פעם אחת' (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics), מגדירים את הפרמטר הזה לערך true. כדי להשתמש בסמנטיקה של 'פעם אחת בדיוק', מגדירים את הפרמטר לערך false. הפרמטר הזה רלוונטי רק אם הערך של useStorageWriteApi הוא true. ערך ברירת המחדל הוא false.
• ‫numStorageWriteApiStreams: כשמשתמשים ב-Storage Write API, מציינים את מספר זרמי הכתיבה. אם useStorageWriteApi הוא true ו-useStorageWriteApiAtLeastOnce הוא false, חובה להגדיר את הפרמטר הזה. ברירת המחדל היא 0.
• ‫storageWriteApiTriggeringFrequencySec: כשמשתמשים ב-Storage Write API, הפרמטר הזה מציין את תדירות ההפעלה בשניות. אם useStorageWriteApi הוא true ו-useStorageWriteApiAtLeastOnce הוא false, חובה להגדיר את הפרמטר הזה.

הפעלת התבנית

המסוף

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

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

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

gcloud

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

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

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Spanner_Change_Streams_to_BigQuery \
    --region REGION_NAME \
    --parameters \
spannerInstanceId=SPANNER_INSTANCE_ID,\
spannerDatabase=SPANNER_DATABASE,\
spannerMetadataInstanceId=SPANNER_METADATA_INSTANCE_ID,\
spannerMetadataDatabase=SPANNER_METADATA_DATABASE,\
spannerChangeStreamName=SPANNER_CHANGE_STREAM,\
bigQueryDataset=BIGQUERY_DATASET

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

• ‫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
• ‫SPANNER_INSTANCE_ID: מזהה מכונת Spanner
• ‫SPANNER_DATABASE: מסד נתונים של Spanner
• ‫SPANNER_METADATA_INSTANCE_ID: מזהה מכונת מטא-נתונים של Spanner
• ‫SPANNER_METADATA_DATABASE: מסד נתונים של מטא-נתונים ב-Spanner
• ‫SPANNER_CHANGE_STREAM: Spanner change stream
• ‫BIGQUERY_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",
      "parameters": {
          "spannerInstanceId": "SPANNER_INSTANCE_ID",
          "spannerDatabase": "SPANNER_DATABASE",
          "spannerMetadataInstanceId": "SPANNER_METADATA_INSTANCE_ID",
          "spannerMetadataDatabase": "SPANNER_METADATA_DATABASE",
          "spannerChangeStreamName": "SPANNER_CHANGE_STREAM",
          "bigQueryDataset": "BIGQUERY_DATASET"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Spanner_Change_Streams_to_BigQuery",
   }
}

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

• ‫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
• ‫SPANNER_INSTANCE_ID: מזהה מכונת Spanner
• ‫SPANNER_DATABASE: מסד נתונים של Spanner
• ‫SPANNER_METADATA_INSTANCE_ID: מזהה מכונת מטא-נתונים של Spanner
• ‫SPANNER_METADATA_DATABASE: מסד נתונים של מטא-נתונים ב-Spanner
• ‫SPANNER_CHANGE_STREAM: Spanner change stream
• ‫BIGQUERY_DATASET: מערך הנתונים ב-BigQuery לפלט של סנכרון שינויים בזרמי נתונים

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

Java

/*
 * Copyright (C) 2022 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.spannerchangestreamstobigquery;

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.Timestamp;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
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.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.SpannerChangeStreamsToBigQueryOptions;
import com.google.cloud.teleport.v2.templates.spannerchangestreamstobigquery.model.Mod;
import com.google.cloud.teleport.v2.templates.spannerchangestreamstobigquery.model.ModColumnType;
import com.google.cloud.teleport.v2.templates.spannerchangestreamstobigquery.schemautils.BigQueryUtils;
import com.google.cloud.teleport.v2.templates.spannerchangestreamstobigquery.schemautils.OptionsUtils;
import com.google.cloud.teleport.v2.transforms.DLQWriteTransform;
import com.google.cloud.teleport.v2.utils.BigQueryIOUtils;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableSet;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
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.StringUtf8Coder;
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.InsertRetryPolicy;
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.DataChangeRecord;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
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.Reshuffle;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

// TODO(haikuo-google): Add integration test.
// TODO(haikuo-google): Add README.
// TODO(haikuo-google): Add stackdriver metrics.
// TODO(haikuo-google): Ideally side input should be used to store schema information and shared
// accross DoFns, but since side input fix is not yet deployed at the moment, we read schema
// information in the beginning of the DoFn as a work around. We should use side input instead when
// it's available.
// TODO(haikuo-google): Test the case where tables or columns are added while the pipeline is
// running.
/**
 * This pipeline ingests {@link DataChangeRecord} from Spanner change stream. The {@link
 * DataChangeRecord} is then broken into {@link Mod}, which converted into {@link TableRow} and
 * inserted into BigQuery table.
 */
@Template(
    name = "Spanner_Change_Streams_to_BigQuery",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Spanner change streams to BigQuery",
    description = {
      "The Cloud Spanner change streams to BigQuery template is a streaming pipeline that streams"
          + " Cloud Spanner data change records and writes them into BigQuery tables using Dataflow"
          + " Runner V2.\n",
      "All change stream watched columns are included in each BigQuery table row, regardless of"
          + " whether they are modified by a Cloud Spanner transaction. Columns not watched are not"
          + " included in the BigQuery row. Any Cloud Spanner change less than the Dataflow"
          + " watermark are either successfully applied to the BigQuery tables or are stored in the"
          + " dead-letter queue for retry. BigQuery rows are inserted out of order compared to the"
          + " original Cloud Spanner commit timestamp ordering.\n",
      "If the necessary BigQuery tables don't exist, the pipeline creates them. Otherwise, existing"
          + " BigQuery tables are used. The schema of existing BigQuery tables must contain the"
          + " corresponding tracked columns of the Cloud Spanner tables and any additional metadata"
          + " columns that are not ignored explicitly by the ignoreFields option. See the"
          + " description of the metadata fields in the following list. Each new BigQuery row"
          + " includes all columns watched by the change stream from its corresponding row in your"
          + " Cloud Spanner table at the change record's timestamp.\n",
      "The following metadata fields are added to BigQuery tables. For more details about these"
          + " fields, see Data change records in \"Change streams partitions, records, and"
          + " queries.\"\n"
          + "- _metadata_spanner_mod_type: The modification type (insert, update, or delete) of the"
          + " Cloud Spanner transaction. Extracted from change stream data change record.\n"
          + "- _metadata_spanner_table_name: The Cloud Spanner table name. Note this field is not"
          + " the metadata table name of the connector.\n"
          + "- _metadata_spanner_commit_timestamp: The Spanner commit timestamp, which is the time"
          + " when a change is committed. Extracted from change stream data change record.\n"
          + "- _metadata_spanner_server_transaction_id: A globally unique string that represents"
          + " the Spanner transaction in which the change was committed. Only use this value in the"
          + " context of processing change stream records. It isn't correlated with the transaction"
          + " ID in Spanner's API. Extracted from change stream data change record.\n"
          + "- _metadata_spanner_record_sequence: The sequence number for the record within the"
          + " Spanner transaction. Sequence numbers are guaranteed to be unique and monotonically"
          + " increasing (but not necessarily contiguous) within a transaction. Extracted from"
          + " change stream data change record.\n"
          + "- _metadata_spanner_is_last_record_in_transaction_in_partition: Indicates whether the"
          + " record is the last record for a Spanner transaction in the current partition."
          + " Extracted from change stream data change record.\n"
          + "- _metadata_spanner_number_of_records_in_transaction: The number of data change"
          + " records that are part of the Spanner transaction across all change stream partitions."
          + " Extracted from change stream data change record.\n"
          + "- _metadata_spanner_number_of_partitions_in_transaction: The number of partitions that"
          + " return data change records for the Spanner transaction. Extracted from change stream"
          + " data change record.\n"
          + "- _metadata_big_query_commit_timestamp: The commit timestamp of when the row is"
          + " inserted into BigQuery.\n",
      "Notes:\n"
          + "- This template does not propagate schema changes from Cloud Spanner to BigQuery."
          + " Because performing a schema change in Cloud Spanner is likely going to break the"
          + " pipeline, you might need to recreate the pipeline after the schema change.\n"
          + "- For OLD_AND_NEW_VALUES and NEW_VALUES value capture types, when the data change"
          + " record contains an UPDATE change, the template needs to do a stale read to Cloud"
          + " Spanner at the commit timestamp of the data change record to retrieve the unchanged"
          + " but watched columns. Make sure to configure your database 'version_retention_period'"
          + " properly for the stale read. For the NEW_ROW value capture type, the template is more"
          + " efficient, because the data change record captures the full new row including columns"
          + " that are not updated in UPDATEs, and the template does not need to do a stale read.\n"
          + "- You can minimize network latency and network transport costs by running the Dataflow"
          + " job from the same region as your Cloud Spanner instance or BigQuery tables. If you"
          + " use sources, sinks, staging file locations, or temporary file locations that are"
          + " located outside of your job's region, your data might be sent across regions. See"
          + " more about Dataflow regional endpoints.\n"
          + "- This template supports all valid Cloud Spanner data types, but if the BigQuery type"
          + " is more precise than the Cloud Spanner type, precision loss might occur during the"
          + " transformation. Specifically:\n"
          + "  - For Cloud Spanner JSON type, the order of the members of an object is"
          + " lexicographically ordered, but there is no such guarantee for BigQuery JSON type.\n"
          + "  - Cloud Spanner supports nanoseconds TIMESTAMP type, BigQuery only supports"
          + " microseconds TIMESTAMP type.\n",
      "Learn more about <a href=\"https://cloud.google.com/spanner/docs/change-streams\">change"
          + " streams</a>, <a"
          + " href=\"https://cloud.google.com/spanner/docs/change-streams/use-dataflow\">how to"
          + " build change streams Dataflow pipelines</a>, and <a"
          + " href=\"https://cloud.google.com/spanner/docs/change-streams/use-dataflow#best_practices\">best"
          + " practices</a>."
    },
    optionsClass = SpannerChangeStreamsToBigQueryOptions.class,
    flexContainerName = "googlecloud-to-googlecloud",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-spanner-change-streams-to-bigquery",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "The Cloud Spanner instance must exist prior to running the pipeline.",
      "The Cloud Spanner database must exist prior to running the pipeline.",
      "The Cloud Spanner metadata instance must exist prior to running the pipeline.",
      "The Cloud Spanner metadata database must exist prior to running the pipeline.",
      "The Cloud Spanner change stream must exist prior to running the pipeline.",
      "The BigQuery dataset must exist prior to running the pipeline."
    },
    streaming = true,
    supportsExactlyOnce = true,
    supportsAtLeastOnce = true)
public final class SpannerChangeStreamsToBigQuery {

  /** 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(SpannerChangeStreamsToBigQuery.class);

  // Max number of deadletter queue retries.
  private static final int DLQ_MAX_RETRIES = 5;

  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) {
    UncaughtExceptionLogger.register();

    LOG.info("Starting to replicate change records from Spanner change streams to BigQuery");

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

    run(options);
  }

  private static void validateOptions(SpannerChangeStreamsToBigQueryOptions options) {
    if (options.getDlqRetryMinutes() <= 0) {
      throw new IllegalArgumentException("dlqRetryMinutes must be positive.");
    }
    if (options
        .getBigQueryChangelogTableNameTemplate()
        .equals(BigQueryUtils.BQ_CHANGELOG_FIELD_NAME_TABLE_NAME)) {
      throw new IllegalArgumentException(
          String.format(
              "bigQueryChangelogTableNameTemplate cannot be set to '{%s}'. This value is reserved"
                  + " for the Cloud Spanner table name.",
              BigQueryUtils.BQ_CHANGELOG_FIELD_NAME_TABLE_NAME));
    }

    BigQueryIOUtils.validateBQStorageApiOptionsStreaming(options);
  }

  private static void setOptions(SpannerChangeStreamsToBigQueryOptions 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<>();
    }
    if (!experiments.contains(USE_RUNNER_V2_EXPERIMENT)) {
      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(SpannerChangeStreamsToBigQueryOptions options) {
    setOptions(options);
    validateOptions(options);

    /**
     * Stages: 1) Read {@link DataChangeRecord} from change stream. 2) Create {@link
     * FailsafeElement} of {@link Mod} JSON and merge from: - {@link DataChangeRecord}. - GCS Dead
     * letter queue. 3) Convert {@link Mod} JSON into {@link TableRow} by reading from Spanner at
     * commit timestamp. 4) Append {@link TableRow} to BigQuery. 5) Write Failures from 2), 3) and
     * 4) to GCS dead letter queue.
     */
    Pipeline pipeline = Pipeline.create(options);
    DeadLetterQueueManager dlqManager = buildDlqManager(options);
    String spannerProjectId = OptionsUtils.getSpannerProjectId(options);

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

    /**
     * There are two types of errors that can occur in this pipeline:
     *
     * <p>1) Error originating from modJsonStringToTableRow. Errors here are either due to pk values
     * missing, a spanner table / column missing in the in-memory map, or some Spanner read error
     * happening in readSpannerRow. We already retry the Spanner read error inline 3 times. Th other
     * types of errors are more likely to be un-retriable.
     *
     * <p>2) Error originating from BigQueryIO.write. BigQuery storage write API already retries all
     * transient errors and outputs more permanent errors.
     *
     * <p>As a result, it is reasonable to write all errors happening in the pipeline directly into
     * the permanent DLQ, since most of the errors are likely to be non-transient.
     */
    if (options.getDisableDlqRetries()) {
      LOG.info(
          "Disabling retries for the DLQ, directly writing into severe DLQ: {}",
          dlqManager.getSevereDlqDirectoryWithDateTime());
      dlqDirectory = dlqManager.getSevereDlqDirectoryWithDateTime();
      tempDlqDirectory = dlqManager.getSevereDlqDirectory() + "tmp/";
    }

    // Retrieve and parse the startTimestamp and endTimestamp.
    Timestamp startTimestamp =
        options.getStartTimestamp().isEmpty()
            ? Timestamp.now()
            : Timestamp.parseTimestamp(options.getStartTimestamp());
    Timestamp endTimestamp =
        options.getEndTimestamp().isEmpty()
            ? Timestamp.MAX_VALUE
            : Timestamp.parseTimestamp(options.getEndTimestamp());

    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withProjectId(spannerProjectId)
            .withInstanceId(options.getSpannerInstanceId())
            .withDatabaseId(options.getSpannerDatabase())
            .withRpcPriority(options.getRpcPriority());
    if (!Strings.isNullOrEmpty(options.getSpannerHost())) {
      spannerConfig =
          spannerConfig.withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()));
    }
    // Propagate database role for fine-grained access control on change stream.
    if (options.getSpannerDatabaseRole() != null) {
      spannerConfig =
          spannerConfig.withDatabaseRole(
              ValueProvider.StaticValueProvider.of(options.getSpannerDatabaseRole()));
    }

    SpannerIO.ReadChangeStream readChangeStream =
        SpannerIO.readChangeStream()
            .withSpannerConfig(spannerConfig)
            .withMetadataInstance(options.getSpannerMetadataInstanceId())
            .withMetadataDatabase(options.getSpannerMetadataDatabase())
            .withChangeStreamName(options.getSpannerChangeStreamName())
            .withInclusiveStartAt(startTimestamp)
            .withInclusiveEndAt(endTimestamp)
            .withRpcPriority(options.getRpcPriority());

    String spannerMetadataTableName = options.getSpannerMetadataTableName();
    if (spannerMetadataTableName != null) {
      readChangeStream = readChangeStream.withMetadataTable(spannerMetadataTableName);
    }

    PCollection<DataChangeRecord> dataChangeRecord =
        pipeline
            .apply("Read from Spanner Change Streams", readChangeStream)
            .apply("Reshuffle DataChangeRecord", Reshuffle.viaRandomKey());

    PCollection<FailsafeElement<String, String>> sourceFailsafeModJson =
        dataChangeRecord
            .apply("DataChangeRecord To Mod JSON", ParDo.of(new DataChangeRecordToModJsonFn()))
            .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 =
        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());

    ImmutableSet.Builder<String> ignoreFieldsBuilder = ImmutableSet.builder();
    for (String ignoreField : options.getIgnoreFields().split(",")) {
      ignoreFieldsBuilder.add(ignoreField);
    }
    ImmutableSet<String> ignoreFields = ignoreFieldsBuilder.build();
    FailsafeModJsonToTableRowTransformer.FailsafeModJsonToTableRowOptions
        failsafeModJsonToTableRowOptions =
            FailsafeModJsonToTableRowTransformer.FailsafeModJsonToTableRowOptions.builder()
                .setSpannerConfig(spannerConfig)
                .setSpannerChangeStream(options.getSpannerChangeStreamName())
                .setIgnoreFields(ignoreFields)
                .setCoder(FAILSAFE_ELEMENT_CODER)
                .setUseStorageWriteApi(options.getUseStorageWriteApi())
                .build();
    FailsafeModJsonToTableRowTransformer.FailsafeModJsonToTableRow failsafeModJsonToTableRow =
        new FailsafeModJsonToTableRowTransformer.FailsafeModJsonToTableRow(
            failsafeModJsonToTableRowOptions);

    PCollectionTuple tableRowTuple =
        failsafeModJson.apply("Mod JSON To TableRow", failsafeModJsonToTableRow);
    // If users pass in the full BigQuery dataset ID (projectId.datasetName), extract the dataset
    // name for the setBigQueryDataset parameter.
    List<String> results = OptionsUtils.processBigQueryProjectAndDataset(options);
    String bigqueryProject = results.get(0);
    String bigqueryDataset = results.get(1);

    BigQueryDynamicDestinations.BigQueryDynamicDestinationsOptions
        bigQueryDynamicDestinationsOptions =
            BigQueryDynamicDestinations.BigQueryDynamicDestinationsOptions.builder()
                .setSpannerConfig(spannerConfig)
                .setChangeStreamName(options.getSpannerChangeStreamName())
                .setIgnoreFields(ignoreFields)
                .setBigQueryProject(bigqueryProject)
                .setBigQueryDataset(bigqueryDataset)
                .setBigQueryTableTemplate(options.getBigQueryChangelogTableNameTemplate())
                .setUseStorageWriteApi(options.getUseStorageWriteApi())
                .build();
    WriteResult writeResult;
    if (!options.getUseStorageWriteApi()) {
      writeResult =
          tableRowTuple
              .get(failsafeModJsonToTableRow.transformOut)
              .apply(
                  "Write To BigQuery",
                  BigQueryIO.<TableRow>write()
                      .to(BigQueryDynamicDestinations.of(bigQueryDynamicDestinationsOptions))
                      .withFormatFunction(element -> removeIntermediateMetadataFields(element))
                      .withFormatRecordOnFailureFunction(element -> element)
                      .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
                      .withWriteDisposition(Write.WriteDisposition.WRITE_APPEND)
                      .withExtendedErrorInfo()
                      .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors()));
    } else {
      writeResult =
          tableRowTuple
              .get(failsafeModJsonToTableRow.transformOut)
              .apply(
                  "Write To BigQuery",
                  BigQueryIO.<TableRow>write()
                      .to(BigQueryDynamicDestinations.of(bigQueryDynamicDestinationsOptions))
                      .withFormatFunction(element -> removeIntermediateMetadataFields(element))
                      .withFormatRecordOnFailureFunction(element -> element)
                      .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
                      .withWriteDisposition(Write.WriteDisposition.WRITE_APPEND)
                      .ignoreUnknownValues()
                      .withAutoSchemaUpdate(true) // only supported when using STORAGE_WRITE_API or
                      // STORAGE_API_AT_LEAST_ONCE.
                      .withExtendedErrorInfo()
                      .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors()));
    }

    PCollection<String> transformDlqJson =
        tableRowTuple
            .get(failsafeModJsonToTableRow.transformDeadLetterOut)
            .apply(
                "Failed Mod JSON During Table Row Transformation",
                MapElements.via(new StringDeadLetterQueueSanitizer()));

    PCollection<String> bqWriteDlqJson =
        BigQueryIOUtils.writeResultToBigQueryInsertErrors(writeResult, options)
            .apply(
                "Failed Mod JSON During BigQuery Writes",
                MapElements.via(new BigQueryDeadLetterQueueSanitizer()));

    PCollectionList.of(transformDlqJson)
        // Generally BigQueryIO storage write retries transient errors, and only more
        // persistent errors make it into DLQ.
        .and(bqWriteDlqJson)
        .apply("Merge Failed Mod JSON From Transform And BigQuery", Flatten.pCollections())
        .apply(
            "Write Failed Mod JSON To DLQ",
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqDirectory)
                .withTmpDirectory(tempDlqDirectory)
                .setIncludePaneInfo(true)
                .build());

    PCollection<FailsafeElement<String, String>> nonRetryableDlqModJsonFailsafe =
        dlqModJson.get(DeadLetterQueueManager.PERMANENT_ERRORS).setCoder(FAILSAFE_ELEMENT_CODER);

    nonRetryableDlqModJsonFailsafe
        .apply(
            "Write Mod JSON With Non-retryable 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 DeadLetterQueueManager buildDlqManager(
      SpannerChangeStreamsToBigQueryOptions options) {
    String tempLocation =
        options.as(DataflowPipelineOptions.class).getTempLocation().endsWith("/")
            ? options.as(DataflowPipelineOptions.class).getTempLocation()
            : options.as(DataflowPipelineOptions.class).getTempLocation() + "/";
    String dlqDirectory =
        options.getDeadLetterQueueDirectory().isEmpty()
            ? tempLocation + "dlq/" + options.getJobName() + "/"
            : options.getDeadLetterQueueDirectory();

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

  /**
   * Remove the following intermediate metadata fields that are not user data from {@link TableRow}:
   * _metadata_error, _metadata_retry_count, _metadata_spanner_original_payload_json.
   */
  private static TableRow removeIntermediateMetadataFields(TableRow tableRow) {
    TableRow cleanTableRow = tableRow.clone();
    Set<String> rowKeys = tableRow.keySet();
    Set<String> metadataFields = BigQueryUtils.getBigQueryIntermediateMetadataFieldNames();

    for (String rowKey : rowKeys) {
      if (metadataFields.contains(rowKey)) {
        cleanTableRow.remove(rowKey);
      } else if (rowKeys.contains("_type_" + rowKey)) {
        cleanTableRow.remove("_type_" + rowKey);
      }
    }

    return cleanTableRow;
  }

  /**
   * DoFn that converts a {@link DataChangeRecord} to multiple {@link Mod} in serialized JSON
   * format.
   */
  static class DataChangeRecordToModJsonFn extends DoFn<DataChangeRecord, String> {

    @ProcessElement
    public void process(@Element DataChangeRecord input, OutputReceiver<String> receiver) {
      for (org.apache.beam.sdk.io.gcp.spanner.changestreams.model.Mod changeStreamsMod :
          input.getMods()) {
        Mod mod =
            new Mod(
                changeStreamsMod.getKeysJson(),
                changeStreamsMod.getNewValuesJson(),
                input.getCommitTimestamp(),
                input.getServerTransactionId(),
                input.isLastRecordInTransactionInPartition(),
                input.getRecordSequence(),
                input.getTableName(),
                input.getRowType().stream().map(ModColumnType::new).collect(Collectors.toList()),
                input.getModType(),
                input.getValueCaptureType(),
                input.getNumberOfRecordsInTransaction(),
                input.getNumberOfPartitionsInTransaction());

        String modJsonString;

        try {
          modJsonString = mod.toJson();
        } catch (IOException e) {
          // Ignore exception and print bad format.
          modJsonString = String.format("\"%s\"", input);
        }
        receiver.output(modJsonString);
      }
    }
  }
}

שיקולים בהפקה

כשמריצים את התבנית Spanner Change Streams to BigQuery בסביבת ייצור, מומלץ לפעול לפי השיטות המומלצות הבאות כדי להבטיח את המהימנות ולמנוע אובדן נתונים:

הקצאת הרשאות לעובדים והתאמה לעומס (scaling)

• הגדרת maxNumWorkers בצורה מתאימה: אם אין מספיק עובדים ב-Dataflow, יכול להיות שצינור הנתונים לא יעמוד בקצב העיבוד של נתוני זרם השינויים. הדבר עלול להוביל לזמן אחזור מוגבר, ובחלק מהתרחישים, לאובדן נתונים פוטנציאלי בגלל פסק זמן של מחבר פנימי ותנאי מירוץ. מספר העובדים המקסימלי צריך להיות מספיק כדי לטפל בשיא של קצב העברת הנתונים לכתיבה ב-Spanner.
• הנחיות לגבי גודל: המספר האופטימלי משתנה בהתאם לעומס העבודה. כדאי לעיין במדריך בנושא סנכרון שינויים בזרמי נתונים ב-Spanner במאמר קביעת הגודל של אשכול Dataflow. עוקבים אחרי הביצועים של צינור המכירות ומבצעים שינויים לפי הצורך. לדוגמה, יכול להיות שחלק מעומסי העבודה עם תפוקה גבוהה ידרשו הגדלה משמעותית של maxNumWorkers (לדוגמה, מ-20 ל-100 או יותר).
• התאמה של התאמה אוטומטית לעומס: התאמה אוטומטית לעומס אופקית שמוגדרת כברירת מחדל ב-Dataflow מבוססת בעיקר על השימוש במעבד. אם יש עיכוב בצינור אבל ניצול המעבד לא גבוה, כדאי לשקול כוונון של התאמה אוטומטית לעומס. הורדת הפרמטר worker_utilization_hint יכולה להפוך את ההתאמה האוטומטית לעומס לרספונסיבית יותר לצווארי בקבוק אחרים. פרטים נוספים זמינים במאמר בנושא התאמה אוטומטית לעומס אופקית.

מעקב והתראות

• עדכניות נתוני הפלט (השהיית המערכת): זהו המדד החשוב ביותר למעקב בתבנית הזו. אם הערך של 'עדכניות נתוני הפלט' (dataflow.googleapis.com/job/system_lag) עולה באופן עקבי, זה מצביע על כך שצינור עיבוד הנתונים לא עומד בקצב של השינויים הנכנסים מ-Spanner .
• הגדרת התראות: מגדירים התראות ב-Cloud Monitoring על המדד Output Data Freshness (רעננות נתוני הפלט). מגדירים בסיס לצינור ומגדירים ספי נמוך וספי גבוה על סמך הדרישות העסקיות שלכם לגבי זמן האחזור של הנתונים. חשוב לבדוק במהירות כל התראה על עלייה מתמשכת במדד הזה. מידע נוסף זמין במאמר מדדים של משימות Dataflow.

פירוש יומנים

• ‫SpannerException: DEADLINE_EXCEEDED: למרות שיש ניסיונות חוזרים במחבר, הודעות com.google.cloud.spanner.SpannerException: DEADLINE_EXCEEDED תכופות ביומני העובדים של Dataflow הן אינדיקציה חזקה לכך שהעובדים מתקשים לקרוא נתונים מסנכרון שינויים בזרמי נתונים של Spanner בזמן. לרוב, זה מעיד על כך שהצינור לא קיבל הקצאה מספקת של משאבי עובדים.

הגדרת עדכונים בזמן אמת ב-Spanner

• תקופת שמירה: חשוב לוודא שretention_period של שינוי הנתונים ב-Spanner מוגדר למשך זמן מספיק כדי לטפל בהאטות או בהפסקות אפשריות בצינור Dataflow. ברירת המחדל היא יום אחד. כדאי להגדיל את משך הזמן הזה ל-3 עד 7 ימים כדי לספק מאגר גדול יותר לצינור העברת הנתונים, כך שהוא יוכל להתאושש ולעבד את כל הנתונים שהצטברו בלי לאבד נתונים בגלל שינויים שתוקפם פג בזרם.

הטמעה של השיטות האלה יכולה לשפר את החוסן (resilience) והביצועים של צינור עיבוד הנתונים של סנכרון שינויים בזרמי נתונים של Spanner ל-BigQuery Dataflow.

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

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