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

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

באמצעות התבנית Spanner change streams to BigQuery, אתם יכולים להזרים רשומות של שינויי נתונים מ-Spanner ולכתוב אותן בטבלאות של BigQuery באמצעות Dataflow Portable Runner. השילוב הזה מאפשר לכם לשלב את הנתונים מ-Spanner עם BigQuery כדי ליהנות מניתוח נתונים בזמן אמת וממחסן נתונים.

סקירה כללית של התבנית

העמודות שלא נצפו לא נכללות בשורה ב-BigQuery. כל השינויים ב-Spanner שקטנים מסימן המים של Dataflow מוחלים בהצלחה על הטבלאות ב-BigQuery או מאוחסנים בתור של הודעות שלא ניתן להעביר (dead-letter queue) לניסיון חוזר. השורה ב-BigQuery מוכנסת לא לפי הסדר, בהשוואה לסדר של חותמת הזמן המקורית של ביצוע ה-commit ב-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, התבנית יעילה יותר כי רשומת שינוי הנתונים לוכדת את השורה החדשה המלאה, כולל עמודות שלא עודכנו בבקשות UPDATE, והתבנית לא צריכה לבצע קריאה של נתונים ישנים.
• כדי לצמצם את זמן האחזור ברשת ואת עלויות התעבורה ברשת, מריצים את משימת Dataflow מאותו אזור שבו נמצאים מופע Spanner או טבלאות ב-BigQuery. אם אתם משתמשים במקורות, ביעדים, במיקומי קבצים זמניים או במיקומי קבצים להעברה שנמצאים מחוץ לאזור של העבודה, יכול להיות שהנתונים שלכם יישלחו בין אזורים. מידע נוסף זמין במאמר בנושא אזורי Dataflow.
• התבנית הזו תומכת בכל סוגי הנתונים התקינים של Spanner. אם הסוג ב-BigQuery מדויק יותר מהסוג ב-Spanner, יכול להיות שתהיה ירידה ברמת הדיוק במהלך הטרנספורמציה. באופן ספציפי:
  • בסוג JSON ב-Spanner, הסדר של חברי אובייקט הוא סדר לקסיקוגרפי, אבל אין ערובה לכך בסוג JSON ב-BigQuery.
  • ‫Spanner תומך בסוג חותמת זמן של ננו-שניות, אבל BigQuery תומך רק בסוג חותמת זמן של מיקרו-שניות.

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

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

• מופע Spanner חייב להתקיים לפני הפעלת צינור הנתונים.
• מסד הנתונים של Spanner צריך להתקיים לפני שמריצים את צינור העיבוד.
• מופע המטא-נתונים של Spanner צריך להתקיים לפני שמריצים את צינור העיבוד.
• מסד הנתונים של המטא-נתונים של Spanner צריך להתקיים לפני שמריצים את צינור העיבוד.
• סנכרון שינויים בזרמי נתונים של Spanner חייב להתקיים לפני שמריצים את הפייפליין.
• נתמכים רק סנכרון שינויים בזרמי נתונים שצופים בטבלאות תחת הסכימה שמוגדרת כברירת מחדל.
• סנכרון שינויים בזרמי נתונים שצופים בטבלאות תחת סכימות אחרות יגרמו ל[שגיאה מסוג table not found](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, בלי קשר להגדרה שלהן כ-NOT NULL בטבלת Spanner. אי אפשר לעדכן את העמודות 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: מספר הדקות בין ניסיונות חוזרים של תור הודעות שלא ניתן להעביר (DLQ). ערך ברירת המחדל הוא 10.
• ‫ignoreFields: רשימה מופרדת בפסיקים של שדות (תלוי אותיות רישיות) להתעלמות. השדות האלה יכולים להיות שדות של טבלאות שנצפו, או שדות מטא-נתונים שנוספו על ידי צינור הנתונים. שדות שהמערכת מתעלמת מהם לא מוכנסים ל-BigQuery. אם מתעלמים מהשדה _metadata_spanner_table_name, המערכת מתעלמת גם מהפרמטר bigQueryChangelogTableNameTemplate. ברירת המחדל היא ריקה.
• ‫disableDlqRetries: האם להשבית ניסיונות חוזרים עבור תור ה-DLQ. ברירת המחדל היא: false.
• ‫spannerChangeStreamTvfNameList: רשימה של שמות של Spanner Change Stream TVF שמופרדים באמצעות נקודה-פסיק, שאותם רוצים לשאול ולאחד. ברירת המחדל היא ריקה.
• ‫spannerDirectedReadOptions: אפשרויות קריאה מכוונות לקריאות של סנכרון שינויים בזרמי נתונים של Cloud Spanner. הערך צריך להיות בפורמט מחרוזת JSON של DirectedReadOptions (למשל, {"includeReplicas":{"replicaSelections":[{"location":"us-central1","type":"READ_ONLY"}]}}). מידע נוסף זמין במאמר בנושא קריאות מונחות (https://docs.cloud.google.com/spanner/docs/directed-reads). ברירת המחדל היא ריקה.
• ‫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/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/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא מקוננת בתיקיית נתונים מתאריך ספציפי המתאימה בדלי – gs://dataflow-templates/
  זהירות: יכול להיות שהתבניות בגרסה העדכנית יעודכנו עם שינויים שיגרמו לבעיות במערכת. כדי למנוע את ההשפעה של שינויי תוכנה שעלולים לגרום לכשל על תהליכי העבודה בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫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
• ‫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/VERSION/flex/Spanner_Change_Streams_to_BigQuery",
   }
}

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

• ‫PROJECT_ID: מזהה Google Cloud הפרויקט שבו רוצים להריץ את משימת Dataflow
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שרוצים להשתמש בה

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

  • latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא מקוננת בתיקיית נתונים מתאריך ספציפי המתאימה בדלי – gs://dataflow-templates/
  זהירות: יכול להיות שהתבניות בגרסה העדכנית יעודכנו עם שינויים שיגרמו לבעיות במערכת. כדי למנוע את ההשפעה של שינויי תוכנה שעלולים לגרום לכשל על תהליכי העבודה בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫LOCATION: האזור שבו רוצים לפרוס את משימת Dataflow, לדוגמה: us-central1
• ‫SPANNER_INSTANCE_ID: מזהה מכונת Spanner
• ‫SPANNER_DATABASE: מסד נתונים של Spanner
• ‫SPANNER_METADATA_INSTANCE_ID: מזהה של מכונת מטא-נתונים של Spanner
• ‫SPANNER_METADATA_DATABASE: מסד נתונים של מטא-נתונים ב-Spanner
• SPANNER_CHANGE_STREAM: סנכרון שינויים בזרמי נתונים של Spanner
• ‫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())
            .withDirectedReadOptions(options.getSpannerDirectedReadOptions());

    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 סנכרון שינויים בזרמי נתונים ל-BigQuery בסביבת ייצור, מומלץ לפעול לפי השיטות המומלצות הבאות כדי להבטיח את המהימנות ולמנוע אובדן נתונים:

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

• הגדרת maxNumWorkers בצורה מתאימה: אם אין מספיק workers ב-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 לעיתים קרובות ביומני ה-worker-ים של Dataflow, זהו אינדיקטור חזק לכך שה-worker-ים מתקשים לקרוא נתונים מסנכרון שינויים בזרמי נתונים של Spanner בזמן. לרוב, זה מעיד על כך שהצינור לא מוקצה מספיק מבחינת משאבי עובדים.

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

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

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

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

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