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

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

הערה: מחבר Spanner change streams לא תומך בהפסקת העבודה של משימה. כדי להפסיק את צינור הנתונים, מבטלים את העבודה במקום לנקז אותו. הפסקת העבודה של משימה באמצעות מחבר של Spanner לשינוי נתונים יכולה לגרום לכך שהמשימה תיתקע במצב של הפסקת עבודה. מידע נוסף זמין במאמר בנושא Draining במסמכי 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 change stream שממנו ייקראו הנתונים.
• ‫bigQueryDataset: מערך הנתונים של BigQuery לפלט של זרמי השינויים.

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

• ‫spannerProjectId: הפרויקט שממנו ייקראו נתוני ה-Change Streams. הערך הזה הוא גם הפרויקט שבו נוצרת טבלת המטא-נתונים של מחבר הזרמים של השינויים. ערך ברירת המחדל של הפרמטר הזה הוא הפרויקט שבו צינור Dataflow פועל.
• ‫spannerDatabaseRole: תפקיד מסד הנתונים ב-Spanner שבו יש להשתמש כשמריצים את התבנית. הפרמטר הזה נדרש רק אם משתמש ה-IAM שמריץ את התבנית הוא משתמש עם בקרת גישה פרטנית. לתפקיד במסד הנתונים צריכה להיות הרשאת SELECT בפיד השינויים והרשאת EXECUTE בפונקציית הקריאה של פיד השינויים. מידע נוסף זמין במאמר בנושא בקרת גישה ברמת גרנולריות גבוהה לזרמי שינויים (https://cloud.google.com/spanner/docs/fgac-change-streams).
• ‫spannerMetadataTableName: השם של טבלת המטא-נתונים של מחבר Spanner change streams שבה רוצים להשתמש. אם לא מספקים את המידע הזה, נוצרת באופן אוטומטי טבלת מטא-נתונים של מחבר 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.
• ‫spannerChangeStreamTvfNameList: רשימה של שמות של Spanner Change Stream TVF, מופרדים באמצעות נקודה-פסיק, שאותם רוצים לשאול ולאחד. ברירת המחדל היא ריק.
• ‫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. עוברים לדף Create job from template (יצירת משימה מתבנית) ב-Dataflow.
כניסה לדף Create job from template
2. בשדה שם המשימה, מזינים שם ייחודי למשימה.
3. אופציונלי: בשדה Regional endpoint (נקודת קצה אזורית), בוחרים ערך מהתפריט הנפתח. אזור ברירת המחדל הוא us-central1.

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

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

gcloud

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

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

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/ \
    --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

כדי להריץ את התבנית באמצעות REST API, שולחים בקשת 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/",
   }
}

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

• ‫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.Arrays;
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);
    }

    String tvfNameListString = options.getSpannerChangeStreamTvfNameList();
    if (tvfNameListString != null && !tvfNameListString.isEmpty()) {
      List<String> tvfNameList =
          Arrays.stream(tvfNameListString.split(";"))
              .filter(name -> !name.trim().isEmpty())
              .collect(Collectors.toList());
      if (!tvfNameList.isEmpty()) {
        readChangeStream = readChangeStream.withTvfNameList(tvfNameList);
      }
    }

    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 מבוסס בעיקר על השימוש במעבד. אם יש עיכובים בצינור אבל ניצול ה-CPU לא גבוה, כדאי לשקול כוונון של שינוי גודל אוטומטי. הורדת הפרמטר 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 ימים כדי לספק מאגר גדול יותר לצינור העברת הנתונים, כך שהוא יוכל להתאושש ולעבד את כל הנתונים שהצטברו בלי לאבד נתונים בגלל שינויים שתוקפם פג בזרם.

הטמעה של השיטות האלה יכולה לשפר את העמידות והביצועים של צינור עיבוד הנתונים שלכם ב-Dataflow מ-Spanner Change Streams אל BigQuery.

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

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