שדרוג של צינור עיבוד נתונים לסטרימינג

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

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

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

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

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

שיטות מומלצות

  • שדרוג גרסת Apache Beam SDK בנפרד משינויים בקוד של צינורות הנתונים.
  • אחרי כל שינוי, כדאי לבדוק את צינור הנתונים לפני שמבצעים עדכונים נוספים.
  • חשוב לשדרג באופן קבוע את גרסת Apache Beam SDK שבה משתמש צינור הנתונים.
  • כדאי להשתמש בשיטות אוטומטיות ככל האפשר, כמו עדכונים במהלך הטיסה או עדכונים אוטומטיים של צינורות מקבילים.
  • כדאי להשתמש בניהול קלט/פלט כשזה אפשרי, כדי ליהנות משדרוגים אוטומטיים של גרסאות המחברים.

ביצוע עדכונים בזמן הטיסה

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

  • העבודה חייבת להשתמש ב-Streaming Engine.
  • המשימה צריכה להיות במצב פעיל.
  • אתם משנים רק את מספר העובדים שהמשימה משתמשת בהם.

מידע נוסף זמין במאמר בנושא הגדרת טווח של שינוי גודל אוטומטי בדף בנושא שינוי גודל אוטומטי אופקי.

הוראות לעדכון של משימה פעילה מופיעות במאמר בנושא עדכון צינור קיים.

יצירה או עדכון אוטומטיים (upsert) של תבניות

כשמפעילים צינורות עיבוד נתונים באמצעות תבנית (תבניות קלאסיות, תבניות Flex,‏ Terraform או Config Connector), אפשר להשתמש בניסוי create_or_update_job כדי ליצור או לעדכן (upsert) פונקציונליות.

כשמציינים create_or_update_job בפרמטר additional_experiments או בדגל additional-experiments:

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

הניסוי הזה מבטל את הצורך לקבוע באופן פרוגרמטי אם להשתמש בפעולת ה-API ליצירה או לעדכון כשמפעילים תבנית.

דוגמאות קוד של Terraform ו-Config Connector שמשתמשות בניסוי הזה מופיעות בקטעים הבאים:

הפעלת משימת החלפה

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

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

הוראות להפעלת עבודת החלפה מופיעות במאמר בנושא הפעלת עבודת החלפה.

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

עצירה והחלפה ידנית

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

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

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

מידע נוסף זמין במאמר הפסקת פעולה של משימת Dataflow. אחרי שמסיימים את המשימה הנוכחית, מתחילים משימה חדשה עם אותו שם.

החלפה אוטומטית של חלקי חילוף

‫Dataflow מספק תמיכה ב-API להשקת עדכון אוטומטי של עצירה והחלפה. תהליך העבודה הזה בסגנון הצהרתי מבטל את השלבים הידניים. מגדירים את העבודה להחלפה, והעבודה החדשה מופעלת ומתאמת את המעבר באופן אוטומטי.

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

שליחת בקשה אוטומטית לעדכון מסוג עצירה והחלפה

כדי להשתמש בתהליך העבודה הזה:

מפעילים בקשה אוטומטית לעדכון מסוג עצירה והחלפה באמצעות אפשרויות השירות הבאות:

Java

אפשרות 1: עדכון באמצעות אותו שם של משימה

--update \
--dataflowServiceOptions="update_strategy_parallel_job_update" \
--dataflowServiceOptions="parallel_replace_job_max_stop_duration=DURATION"
  • כדי לבצע עדכון אוטומטי של עצירה והחלפה באמצעות אותו שם, משתמשים בדגל --update ובאפשרות update_strategy_parallel_job_update.
  • כדי לבצע עדכון במקום, משתמשים ב-update_strategy_in_place_update.

אפשרות 2: עדכון באמצעות שם משרה אחר

--dataflowServiceOptions="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflowServiceOptions="parallel_replace_job_max_stop_duration=DURATION"
  • כדי לציין את העבודה הישנה באמצעות מזהה במקום שם העבודה, משתמשים ב---dataflowServiceOptions="parallel_replace_job_id=OLD_JOB_ID".
  • אם מציינים שם חדש למשימה ומשתמשים בדגל --update, ‏ Dataflow מחפש משימה קיימת עם השם החדש, וזה גורם לשגיאה.

אופציונלי: השבתת ביטול אוטומטי

ביטול אוטומטי מופעל כברירת מחדל כשמציינים את האפשרות parallel_replace_job_max_stop_duration. כדי להשבית את הביטול האוטומטי, מגדירים את האפשרות parallel_replace_job_cancel_on_drain_timeout לערך false.

--dataflowServiceOptions="parallel_replace_job_cancel_on_drain_timeout=false"

אם משביתים את הביטול האוטומטי והמשימה הישנה נתקעת במצב של ניקוי, גם המשימה הישנה וגם המשימה החדשה ממשיכות לפעול במקביל.

Python

אפשרות 1: עדכון באמצעות אותו שם של משימה

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"
  • כדי לבצע עדכון אוטומטי של עצירה והחלפה באמצעות אותו שם, משתמשים בדגל --update ובאפשרות update_strategy_parallel_job_update.
  • כדי לבצע עדכון במקום, משתמשים ב-update_strategy_in_place_update.

אפשרות 2: עדכון באמצעות שם משרה אחר

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"
  • כדי לציין את העבודה הישנה באמצעות מזהה במקום שם העבודה, משתמשים ב---dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
  • אם מציינים שם חדש למשימה ומשתמשים בדגל --update, ‏ Dataflow מחפש משימה קיימת עם השם החדש, וזה גורם לשגיאה.

אופציונלי: השבתת ביטול אוטומטי

ביטול אוטומטי מופעל כברירת מחדל כשמציינים את האפשרות parallel_replace_job_max_stop_duration. כדי להשבית את הביטול האוטומטי, מגדירים את האפשרות parallel_replace_job_cancel_on_drain_timeout לערך false.

--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=false"

אם משביתים את הביטול האוטומטי והמשימה הישנה נתקעת במצב של ניקוז, גם המשימה הישנה וגם המשימה החדשה ממשיכות לפעול במקביל.

המשך

אפשרות 1: עדכון באמצעות אותו שם של משימה

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"
  • כדי לבצע עדכון אוטומטי של עצירה והחלפה באמצעות אותו שם, משתמשים בדגל --update ובאפשרות update_strategy_parallel_job_update.
  • כדי לבצע עדכון במקום, משתמשים ב-update_strategy_in_place_update.

אפשרות 2: עדכון באמצעות שם משרה אחר

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"
  • כדי לציין את העבודה הישנה באמצעות מזהה במקום שם העבודה, משתמשים ב---dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
  • אם מציינים שם חדש למשימה ומשתמשים בדגל --update, ‏ Dataflow מחפש משימה קיימת עם השם החדש, וזה גורם לשגיאה.

אופציונלי: השבתת ביטול אוטומטי

ביטול אוטומטי מופעל כברירת מחדל כשמציינים את האפשרות parallel_replace_job_max_stop_duration. כדי להשבית את הביטול האוטומטי, מגדירים את האפשרות parallel_replace_job_cancel_on_drain_timeout לערך false.

--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=false"

אם משביתים את הביטול האוטומטי והמשימה הישנה נתקעת במצב של ניקוי, גם המשימה הישנה וגם המשימה החדשה ממשיכות לפעול במקביל.

gcloud

אפשרות 1: עדכון באמצעות אותו שם של משימה

--update \
--additional-experiments="update_strategy_parallel_job_update" \
--additional-experiments="parallel_replace_job_max_stop_duration=DURATION"
  • כדי לבצע עדכון אוטומטי של עצירה והחלפה באמצעות אותו שם, משתמשים בדגל --update ובאפשרות update_strategy_parallel_job_update.
  • כדי לבצע עדכון במקום, משתמשים ב-update_strategy_in_place_update.

אפשרות 2: עדכון באמצעות שם משרה אחר

--additional-experiments="parallel_replace_job_name=OLD_JOB_NAME" \
--additional-experiments="parallel_replace_job_max_stop_duration=DURATION"
  • כדי לציין את העבודה הישנה באמצעות מזהה במקום שם העבודה, משתמשים ב---additional-experiments="parallel_replace_job_id=OLD_JOB_ID".
  • אם מציינים שם חדש למשימה ומשתמשים בדגל --update, ‏ Dataflow מחפש משימה קיימת עם השם החדש, וזה גורם לשגיאה.

אופציונלי: השבתת ביטול אוטומטי

ביטול אוטומטי מופעל כברירת מחדל כשמציינים את האפשרות parallel_replace_job_max_stop_duration. כדי להשבית את הביטול האוטומטי, מגדירים את האפשרות parallel_replace_job_cancel_on_drain_timeout לערך false.

--additional-experiments="parallel_replace_job_cancel_on_drain_timeout=false"

אם משביתים את הביטול האוטומטי והמשימה הישנה נתקעת במצב של ניקוי, גם המשימה הישנה וגם המשימה החדשה ממשיכות לפעול במקביל.

אופציונלי: upsert (יצירה או עדכון של משימה)

כדי להפעיל התנהגות של upsert (יצירה או עדכון של משימה):

--additional-experiments="create_or_update_job"

Terraform

additional_experiments = [
  "parallel_replace_job_max_stop_duration=DURATION",
  "parallel_replace_job_cancel_on_drain_timeout=true",
  "update_strategy_parallel_job_update",
  "parallel_replace_job_preallocate_compute_resources=true",
  "create_or_update_job"
]

Config Connector 

metadata:
  annotations:
    # Force to use only direct controller. Multiple controllers can cause a Dataflow job to enter into a continuous update loop.
    # https://docs.cloud.google.com/config-connector/docs/concepts/controller-types#underlying-controller-types
    alpha.cnrm.cloud.google.com/reconciler: direct
    # Optional but recommended: Dataflow batch jobs do not support the drain operation. But for Dataflow streaming jobs, prefer "drain" over "cancel" as an on-delete option.
    cnrm.cloud.google.com/on-delete: drain
    # Optional but recommended: Set deletion-policy to "abandon" to avoid accidental deletion, this will ignore the on-delete option.
    cnrm.cloud.google.com/deletion-policy: abandon
spec:
  ...
  additionalExperiments:
    - "parallel_replace_job_max_stop_duration=DURATION"
    - "parallel_replace_job_cancel_on_drain_timeout=true"
    - "update_strategy_parallel_job_update"
    - "parallel_replace_job_preallocate_compute_resources=true"
    - "create_or_update_job"

מחליפים את המשתנים הבאים:

  • כדי לזהות את המשרה שרוצים להחליף, צריך לציין את parallel_replace_job_name או את parallel_replace_job_id:
    • ‫OLD_JOB_NAME: שם המשימה שרוצים להחליף.
    • ‫OLD_JOB_ID: המזהה של המשרה שרוצים להחליף.
  • כדי להפעיל את התכונה 'החלפה אוטומטית של מילות מפתח', צריך לציין את הערך parallel_replace_job_max_stop_duration:
    • ‫DURATION: משך הזמן המקסימלי שהמשימה החדשה ממתינה עד לסיום הניקוז של המשימה הקודמת. המשך צריך להיות בפורמט של מחרוזת שמסתיימת ב-s, ב-m או ב-h (לדוגמה, 30m, 1h).
  • אל תגדירו את האפשרות parallel_replace_job_min_parallel_pipelines_duration כשמשתמשים בתהליך העבודה הזה. הגדרת האפשרות הזו מפעילה במקום זאת את זרימת העבודה של עדכונים אוטומטיים של צינורות מקבילים.
  • אופציונלי: מגדירים את האפשרות parallel_replace_job_cancel_on_drain_timeout. הביטול האוטומטי מופעל כברירת מחדל (true) כשמגדירים את האפשרות parallel_replace_job_max_stop_duration, ולכן לא צריך להגדיר את האפשרות הזו באופן מפורש כדי להפעיל אותה.
    • כדי לשמור על התנהגות ברירת המחדל, משמיטים את האפשרות הזו או מגדירים אותה לערך true.
    • כדי להשבית את הביטול האוטומטי, מגדירים את האפשרות הזו לערך false. אם מגדירים את האפשרות הזו לערך false והמשימה הישנה נתקעת במצב draining, גם המשימה הישנה וגם המשימה החדשה ממשיכות לפעול במקביל.
  • אופציונלי: מגדירים parallel_replace_job_preallocate_compute_resources הגדרות:
    • ההגדרה מציינת אם עובדים מוקצים מראש לעבודה החדשה בזמן שהעבודה הישנה מתרוקנת. הערכים האפשריים: true (ברירת מחדל) או false. ב-Terraform וב-Config Connector, מומלץ להגדיר את האפשרות הזו ל-true כדי למנוע מצב של פסק זמן בהקצאת משאבים. כשהערך של parallel_replace_job_preallocate_compute_resources הוא false, המשימה החדשה נשארת במצב 'בהמתנה' עד שהמשימה הישנה מסתיימת.

עיבוד מחדש של הודעות באמצעות Pub/Sub Snapshot ו-Seek

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

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

הנה תהליך עבודה מומלץ באמצעות ה-CLI של gcloud לשימוש ב-Pub/Sub Seek עם צינורות Dataflow בחלון טרמינל:

  1. כדי ליצור snapshot של המינוי, משתמשים בפקודה gcloud pubsub snapshots create:

    gcloud pubsub snapshots create SNAPSHOT_NAME --subscription=PIPELINE_SUBSCRIPTION_NAME

  2. כדי לנקז את הנתונים מהצינור או לבטל אותו, משתמשים בפקודה gcloud dataflow jobs drain או בפקודה gcloud dataflow jobs cancel:

    gcloud dataflow jobs drain JOB_ID

    או

    gcloud dataflow jobs cancel JOB_ID

  3. כדי לעבור לתמונת המצב, משתמשים בפקודה gcloud pubsub subscriptions seek:

    gcloud pubsub subscriptions seek SNAPSHOT_NAME

  4. פריסת צינור חדש שמשתמש במינוי.

הפעלת צינורות במקביל

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

סקירה כללית על צינורות עיבוד נתונים מקבילים

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

התרשים הבא ממחיש את התהליך הזה.

הצינור B חופף לצינור B בחלון זמן של 5 דקות.

בתרשים, פייפליין ב' הוא העבודה המעודכנת שמחליפה את פייפליין א'. הערך t הוא חותמת הזמן של החלון המוקדם ביותר שPipeline B עיבד. הערך w הוא סימן המים של Pipeline A. לצורך פשטות, נניח שיש סימן מים מושלם ללא נתונים שמתקבלים באיחור. זמן העיבוד והזמן הכולל מוצגים בציר האופקי. בשני צינורות הנתונים נעשה שימוש בחלונות קבועים (מתגלגלים) של חמש דקות. התוצאות מופעלות אחרי שהסימן הגרפי עובר את סוף כל חלון.

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

מגבלות

יש מגבלות על שימוש בעדכונים אוטומטיים או ידניים של צינורות מקבילים:

  • עדכונים אוטומטיים בלבד: המשימה המקבילה החדשה חייבת להיות משימה של Streaming Engine.
  • אסור להריץ משימות בו-זמנית עם אותו שם. עם זאת, כשמבצעים עצירה והחלפה אוטומטיות או עדכון מקביל של צינורות באמצעות אותו שם של משימה, אפשר להשתמש שוב בשם המשימה. במקרה כזה, העבודה החדשה צריכה להתחיל לפחות שתי דקות אחרי תחילת העבודה הקודמת. ההגבלה הזו מונעת עדכונים מקבילים מרובים מניסיונות חוזרים של ספריות לקוח או מקריאות מרוחקות של פרוצדורות לא עדכניות.
  • הפעלת שני צינורות במקביל על אותו קלט עלולה להוביל לכפילות בנתונים, לצבירות חלקיות ולבעיות פוטנציאליות בסדר כשמכניסים נתונים ליעד. המערכת במורד הזרם צריכה להיות מתוכננת כך שתצפה לתוצאות האלה ותנהל אותן.
  • כשקוראים ממקור Pub/Sub, לא מומלץ להשתמש באותו מינוי לכמה צינורות, כי זה עלול לגרום לבעיות בדיוק הנתונים. עם זאת, בתרחישי שימוש מסוימים, כמו צינורות ETL (חילוץ, טרנספורמציה, טעינה), שימוש באותו מינוי בשני צינורות עשוי לצמצם כפילויות. בעיות בסקיילינג אוטומטי עלולות לקרות בכל פעם שמזינים ערך שונה מאפס לפרק הזמן החופף. כדי לפתור את הבעיה, אפשר להשתמש בתכונה לעדכון משימות בזמן שהן מתבצעות. מידע נוסף זמין במאמר בנושא התאמה מדויקת של שינוי גודל אוטומטי של צינורות סטרימינג ב-Pub/Sub.
  • ב-Apache Kafka, אפשר למזער כפילויות על ידי הפעלת offset committing ב-Kafka. כדי להפעיל את האפשרות 'שמירת היסטוריית ההתקדמות' ב-Kafka, אפשר לעיין במאמר שמירת היסטוריית ההתקדמות ב-Kafka.

עדכונים אוטומטיים של צינורות לעיבוד נתונים במקביל

‫Dataflow מספק תמיכה ב-API להפעלת עבודת החלפה מקבילה. ממשק ה-API הזה בסגנון דקלרטיבי מבצע הפשטה של העבודה הידנית שנדרשת להרצת שלבים פרוצדורליים. מצהירים על המשימה שרוצים לעדכן, ואז משימה חדשה פועלת במקביל למשימה הישנה. אחרי שהמשימה החדשה תפעל למשך הזמן שתציינו, המשימה הישנה תתרוקן. התכונה הזו מבטלת את ההשהיות בעיבוד במהלך עדכונים. בנוסף, הוא מפחית את המאמץ התפעולי שנדרש לעדכון צינורות (pipelines) לא תואמים.

שיטת העדכון הזו מתאימה במיוחד לצינורות עיבוד נתונים שיכולים להכיל כפילויות או צבירות חלקיות, ושלא נדרש בהם סדר קפדני בזמן הוספת הנתונים. הוא מתאים במיוחד לצינורות ETL, וגם לצינורות שמשתמשים במצב סטרימינג של לפחות פעם אחתובטרנספורמציה Redistribute עם ההגדרה allow duplicates (התרת כפילויות) שמוגדרת לערך true.

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

אפשר להשתמש באפשרויות השירות הבאות לעדכונים אוטומטיים של צינורות מקבילים:

אפשרות שירות אופציונלי או חובה תיאור תלות או החרגות
update_strategy_parallel_job_update חובה (אפשרות 1: עדכון באמצעות אותו שם משימה) פקודה לביצוע עדכון מקביל, שמריצה את שני צינורות העיבוד בו-זמנית כדי למזער את זמן ההשבתה, כשמעדכנים תחת אותו שם של משימה. חובה להגדיר אותו לצד הדגל --update והדגל parallel_replace_job_min_parallel_pipelines_duration.
update_strategy_in_place_update אופציונלי חלופה לעדכון מקביל. מבצע עדכון רגיל של משימה במקום. צריך להגדיר אותו לצד הדגל --update.

בלעדיות הדדית עם update_strategy_parallel_job_update.

אם האפשרות הזו מוגדרת, המערכת מתעלמת מאפשרויות אחרות שקשורות למשימות מקבילות.
parallel_replace_job_min_parallel_pipelines_duration חובה מציין את משך הזמן המינימלי שבו שני צינורות העיבוד פועלים בו-זמנית. אחרי פרק הזמן הזה, אות ניקוז נשלח לעבודה הישנה. הערכים הקבילים הם בין 0s (מומלץ לחפיפה אפס) לבין 744h (31 ימים). צריך לשלב את ההגדרה הזו עם דרך לטרגט את המשרה הישנה. אחת מהאפשרויות הבאות:
  • אפשרות 1 – שימוש באותו שם משרה: update_strategy_parallel_job_update, או
  • אפשרות 2 – שימוש בשם אחר של המשימה: ‫parallel_replace_job_name (או לחלופין ‫parallel_replace_job_id כדי לזהות את המשימה)
‫parallel_replace_job_name או parallel_replace_job_id (צריך לבחור באחת מהאפשרויות) חובה (אפשרות 2: עדכון באמצעות שם משרה אחר) מזהה את המשרה הישנה לפי שם או מזהה, כדי להחליף אותה במהלך עדכון של שם אחר. נדרשת הגדרה של parallel_replace_job_min_parallel_pipelines_duration.

אל תשתמשו בדגל --update או באפשרות parallel_replace_job_id עם האפשרות הזו.
parallel_replace_job_max_stop_duration אופציונלי המשך המקסימלי של העבודה הישנה לפני הפעלה של ביטול אוטומטי. לדוגמה, 30m או 1h. כדי להשתמש בתכונה הזו, צריך להגדיר תהליך עבודה לעדכון מקביל (אפשרות 1 או אפשרות 2).
parallel_replace_job_cancel_on_drain_timeout אופציונלי

אם מוגדר משך מקסימלי של עצירה, ברירת המחדל היא true.

 אפשרות בוליאנית שמציינת אם לבטל את העבודה הישנה אם משך הניקוז שלה חורג מ-parallel_replace_job_max_stop_duration. בשימוש בשילוב עם parallel_replace_job_max_stop_duration.

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

שליחת בקשה אוטומטית לעדכון של צינור מקביל לעיבוד נתונים

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

Java

אפשרות 1: עדכון באמצעות אותו שם של משימה

--update \
--dataflowServiceOptions="update_strategy_parallel_job_update" \
--dataflowServiceOptions="parallel_replace_job_min_parallel_pipelines_duration=DURATION"
  • כדי לבצע עדכון מקביל באמצעות אותו שם, משתמשים בדגל --update ובאפשרות update_strategy_parallel_job_update.
  • כדי לבצע עדכון במקום בלי להסיר אפשרויות שקשורות למשימות מקבילות, משתמשים ב-update_strategy_in_place_update במקום ב-update_strategy_parallel_job_update.

אפשרות 2: עדכון באמצעות שם משרה אחר

--dataflowServiceOptions="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflowServiceOptions="parallel_replace_job_min_parallel_pipelines_duration=DURATION"
  • כדי לציין את העבודה הישנה באמצעות מזהה במקום שם העבודה, משתמשים ב---dataflowServiceOptions="parallel_replace_job_id=OLD_JOB_ID".
  • אם מציינים שם חדש למשימה ומשתמשים בדגל --update, ‏ Dataflow מחפש משימה קיימת עם השם החדש, וזה גורם לשגיאה.

אופציונלי: הגדרת פסק זמן לסיום התהליך וביטול אוטומטי

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

--dataflowServiceOptions="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--dataflowServiceOptions="parallel_replace_job_cancel_on_drain_timeout=true"

אם משביתים את הביטול האוטומטי והמשימה הישנה נתקעת במצב של ניקוי, גם המשימה הישנה וגם המשימה החדשה ממשיכות לפעול במקביל.

Python

אפשרות 1: עדכון באמצעות אותו שם של משימה

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"
  • כדי לבצע עדכון מקביל באמצעות אותו שם, משתמשים בדגל --update ובאפשרות update_strategy_parallel_job_update.
  • כדי לבצע עדכון במקום בלי להסיר אפשרויות שקשורות למשימות מקבילות, משתמשים ב-update_strategy_in_place_update במקום ב-update_strategy_parallel_job_update.

אפשרות 2: עדכון באמצעות שם משרה אחר

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"
  • כדי לציין את העבודה הישנה באמצעות מזהה במקום שם העבודה, משתמשים ב---dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
  • אם מציינים שם חדש למשימה ומשתמשים בדגל --update, ‏ Dataflow מחפש משימה קיימת עם השם החדש, וזה גורם לשגיאה.

אופציונלי: הגדרת פסק זמן לסיום התהליך וביטול אוטומטי

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

--dataflow_service_options="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=true"

אם משביתים את הביטול האוטומטי והמשימה הישנה נתקעת במצב של ניקוי, גם המשימה הישנה וגם המשימה החדשה ממשיכות לפעול במקביל.

המשך

אפשרות 1: עדכון באמצעות אותו שם של משימה

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"
  • כדי לבצע עדכון מקביל באמצעות אותו שם, משתמשים בדגל --update ובאפשרות update_strategy_parallel_job_update.
  • כדי לבצע עדכון במקום בלי להסיר אפשרויות שקשורות למשימות מקבילות, משתמשים ב-update_strategy_in_place_update במקום ב-update_strategy_parallel_job_update.

אפשרות 2: עדכון באמצעות שם משרה אחר

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"
  • כדי לציין את העבודה הישנה באמצעות מזהה במקום שם העבודה, משתמשים ב---dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
  • אם מציינים שם חדש למשימה ומשתמשים בדגל --update, ‏ Dataflow מחפש משימה קיימת עם השם החדש, וזה גורם לשגיאה.

אופציונלי: הגדרת פסק זמן לסיום התהליך וביטול אוטומטי

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

--dataflow_service_options="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=true"

אם משביתים את הביטול האוטומטי והמשימה הישנה נתקעת במצב של ניקוי, גם המשימה הישנה וגם המשימה החדשה ממשיכות לפעול במקביל.

gcloud

אפשרות 1: עדכון באמצעות אותו שם של משימה

--update \
--additional-experiments="update_strategy_parallel_job_update" \
--additional-experiments="parallel_replace_job_min_parallel_pipelines_duration=DURATION"
  • כדי לבצע עדכון מקביל באמצעות אותו שם, משתמשים בדגל --update ובאפשרות update_strategy_parallel_job_update.
  • כדי לבצע עדכון במקום בלי להסיר אפשרויות שקשורות למשימות מקבילות, משתמשים ב-update_strategy_in_place_update במקום ב-update_strategy_parallel_job_update.

אפשרות 2: עדכון באמצעות שם משרה אחר

--additional-experiments="parallel_replace_job_name=OLD_JOB_NAME" \
--additional-experiments="parallel_replace_job_min_parallel_pipelines_duration=DURATION"
  • כדי לציין את העבודה הישנה באמצעות מזהה במקום שם העבודה, משתמשים ב---additional-experiments="parallel_replace_job_id=OLD_JOB_ID".
  • אם מציינים שם חדש למשימה ומשתמשים בדגל --update,‏ Dataflow מחפש משימה קיימת עם השם החדש, וזה גורם לשגיאה.

אופציונלי: הגדרת פסק זמן לסיום התהליך וביטול אוטומטי

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

--additional-experiments="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--additional-experiments="parallel_replace_job_cancel_on_drain_timeout=true"

אם משביתים את הביטול האוטומטי והמשימה הישנה נתקעת במצב של ניקוי, גם המשימה הישנה וגם המשימה החדשה ממשיכות לפעול במקביל.

אופציונלי: upsert (יצירה או עדכון של משימה)

כדי להפעיל התנהגות של upsert (יצירה או עדכון של משימה):

--additional-experiments="create_or_update_job"

Terraform 

additional_experiments = [
  "parallel_replace_job_min_parallel_pipelines_duration=DURATION",
  "parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION",
  "update_strategy_parallel_job_update",
  "create_or_update_job"
]

Config Connector 

metadata:
  annotations:
    # Force to use only direct controller. Multiple controllers can cause a Dataflow job to enter into a continuous update loop.
    # https://docs.cloud.google.com/config-connector/docs/concepts/controller-types#underlying-controller-types
    alpha.cnrm.cloud.google.com/reconciler: direct
    # Optional but recommended: Dataflow batch jobs do not support the drain operation. But for Dataflow streaming jobs, prefer "drain" over "cancel" as an on-delete option.
    cnrm.cloud.google.com/on-delete: drain
    # Optional but recommended: Set deletion-policy to "abandon" to avoid accidental deletion, this will ignore the on-delete option.
    cnrm.cloud.google.com/deletion-policy: abandon
spec:
  ...
  additionalExperiments:
    - "parallel_replace_job_min_parallel_pipelines_duration=DURATION"
    - "parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION"
    - "update_strategy_parallel_job_update"
    - "create_or_update_job"

מחליפים את המשתנים הבאים:

  • אם אתם מעדכנים באמצעות שם תפקיד שונה (אפשרות 2), אתם צריכים לספק את parallel_replace_job_name או את parallel_replace_job_id כדי לזהות את התפקיד שרוצים להחליף. אי אפשר לעדכן באמצעות שם משימה שונה ב-Terraform או ב-Config Connector.
    • ‫OLD_JOB_NAME: שם המשימה שרוצים להחליף.
    • ‫OLD_JOB_ID: המזהה של המשרה שרוצים להחליף.

  • ‫DURATION: משך הזמן המינימלי ששני צינורות העיבוד פועלים במקביל, כמספר שלם או כמספר נקודה צפה (floating-point). מומלץ להגדיר משך זמן של 0s כדי שלא יהיה חפיפה בין המבצעים. אחרי שמשך הזמן הזה חולף, אות ניקוז נשלח לעבודה הישנה.

    משך הזמן צריך להיות בין 0 שניות (0s) ל-31 ימים (744h). משתמשים ב-s,‏ m ו-h כדי לציין שניות, דקות ושעות. לדוגמה, 10m הוא 10 דקות.

  • ‫DRAIN_TIMEOUT_DURATION: אופציונלי. הזמן המקסימלי שצריך לחכות עד שהעבודה הקודמת תסתיים לפני שהביטול האוטומטי מופעל. משך הזמן צריך להיות בפורמט של מחרוזת שמסתיימת ב-s, ב-m או ב-h (לדוגמה, 30m, 1h).

  • ‫parallel_replace_job_cancel_on_drain_timeout: אופציונלי. האם לבטל את המשימה הקודמת אם היא לא מסתיימת לפני משך הזמן המקסימלי להפסקה. אם מציינים משך זמן קצוב לתהליך ההתרוקנות, ברירת המחדל היא true. כדי להשבית את הביטול האוטומטי, מגדירים את האפשרות הזו ל-false. אם מגדירים את האפשרות הזו לערך false והמשימה הישנה נתקעת במצב draining, גם המשימה הישנה וגם המשימה החדשה ממשיכות לפעול במקביל.

כשמפעילים את המשימה החדשה, Dataflow מחכה שכל העובדים יוקצו לפני שהוא מתחיל לעבד את הנתונים. כדי לעקוב אחרי סטטוס הפריסה, בודקים את יומני המשימות של Dataflow.

הפעלה ידנית של צינורות מקבילים

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

טיפול בפלט משוכפל

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

  1. במצב ההתחלתי, צינור הנתונים הקיים בסטרימינג (Pipeline A) פועל וקורא הודעות מנושא ב-Pub/Sub‏ (Topic) באמצעות מינוי (Subscription A). התוצאות נכתבות לטבלה ב-BigQuery (טבלה א'). התוצאות נצרכות דרך תצוגה ב-BigQuery, שמשמשת כחזית להסתרת שינויים בטבלה הבסיסית. התהליך הזה הוא יישום של שיטת עיצוב שנקראת תבנית חזית. התרשים הבא מציג את המצב ההתחלתי.

    פייפליין אחד עם מינוי אחד, וכתיבה לטבלה ב-BigQuery אחת.

  2. יוצרים מינוי חדש (מינוי ב') לצינור המעודכן. פורסים את צינור עיבוד הנתונים המעודכן (Pipeline B), שקורא מנושא Pub/Sub ‏ (Topic) באמצעות Subscription B וכותב לטבלת BigQuery נפרדת (Table B). התרשים הבא ממחיש את התהליך הזה.

    שני צינורות, לכל אחד מהם מינוי אחד. כל צינור כתיבה כותב לטבלת BigQuery נפרדת. תצוגת חזית קוראת משתי הטבלאות.

    בשלב הזה, Pipeline A ו-Pipeline B פועלים במקביל וכותבים את התוצאות לטבלאות נפרדות. אתם מתעדים את הזמן t כחותמת הזמן של החלון המוקדם ביותר שמעובד על ידי Pipeline B.

  3. כשהסימן של Pipeline A חורג מהזמן t, צריך לנקז את Pipeline A. כשמרוקנים את צינור הנתונים, כל החלונות הפתוחים נסגרים והעיבוד של הנתונים שנמצאים בתהליך מסתיים. אם צינור הנתונים מכיל חלונות וחשובים לכם חלונות מלאים (בהנחה שאין נתונים מאוחרים), לפני שמרוקנים את צינור הנתונים א', מפעילים את שני צינורות הנתונים עד שמתקבלים חלונות חופפים מלאים. מפסיקים את משימת הסטרימינג של Pipeline A אחרי שכל הנתונים בתהליך העברה מעובדים ונכתבים ב-Table A. התרשים הבא מציג את השלב הזה.

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

  4. בשלב הזה, פועל רק Pipeline B. אפשר לשאול שאילתה מתצוגה של BigQuery (תצוגת חזית), שמשמשת כחזית לטבלה א' ולטבלה ב'. לגבי שורות עם חותמת זמן זהה בשתי הטבלאות, צריך להגדיר את התצוגה כך שהיא תחזיר את השורות מטבלה ב', או, אם השורות לא קיימות בטבלה ב', תחזור לטבלה א'. בתרשים הבא מוצגת התצוגה (Façade View) שקוראת מטבלה א' ומטבלה ב'.

    צינור עיבוד הנתונים א' נעלם, ורק צינור עיבוד הנתונים ב' פועל.

    בשלב הזה, אפשר למחוק את מינוי א'.

אם מתגלות בעיות בפריסה של צינור חדש, צינורות מקבילים יכולים לפשט את החזרה לגרסה הקודמת. בדוגמה הזו, יכול להיות שתרצו להמשיך להפעיל את צינור עיבוד הנתונים א' בזמן שאתם עוקבים אחרי הפעולה התקינה של צינור עיבוד הנתונים ב'. אם מתעוררות בעיות בPipeline B, אפשר לחזור לPipeline A.

טיפול במוטציות של סכימה

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

נניח שיש צינור שקורא הודעות שמכילות מטען ייעודי (payload) של JSON מנושא Pub/Sub. הצינור ממיר כל הודעה למופע של TableRow ואז כותב את השורות לטבלה ב-BigQuery. הסכימה של טבלת הפלט דומה להודעות שעוברות עיבוד בצינור. בתרשים הבא, הסכימה נקראת סכימה א'.

צינור (Pipeline) שקורא מינוי וכותב לטבלת פלט ב-BigQuery באמצעות סכימה א'.

עם הזמן, סכמת ההודעות עשויה להשתנות בדרכים משמעותיות. לדוגמה, שדות נוספים, מוסרים או מוחלפים. ‫Schema A מתפתח לסכימה חדשה. בדיון שבהמשך, הסכימה החדשה נקראת Schema B. במקרה הזה, צריך לעדכן את Pipeline A, וסכימת טבלת הפלט צריכה לתמוך ב-Schema B.

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

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

בתרשים הבא מוצג עדכון של זרימת הצינור הקודמת, שכולל טבלת ביניים (Staging Table A), טבלה ראשית ותצוגת חזית.

צינור שקורא מינוי וכותב לטבלת ביניים ב-BigQuery. בטבלה השנייה (הראשית) יש פלט מגרסה קודמת של הסכימה. תצוגת חזית קוראת נתונים גם מטבלת הביניים וגם מהטבלה הראשית.

בתהליך המתוקן, Pipeline A מעבד הודעות שמשתמשות ב-Schema A וכותב את הפלט ל-Staging Table A, שיש לו סכימה תואמת. הטבלה הראשית מכילה נתונים היסטוריים שנכתבו על ידי גרסאות קודמות של צינור הנתונים, וגם תוצאות שמתבצע מיזוג שלהן באופן תקופתי מטבלת הביניים. הצרכנים יכולים לשאול שאלות לגבי נתונים עדכניים, כולל נתונים היסטוריים ונתונים בזמן אמת, באמצעות תצוגת ה-facade.

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

הצינור המעודכן כותב לטבלת ביניים נוספת (Staging Table B) שמשתמשת ב-Schema B. אתם יכולים להשתמש בתהליך עבודה מתואם כדי ליצור את טבלת הביניים החדשה לפני שתעדכנו את צינור הנתונים. מעדכנים את תצוגת החזית כך שתכלול תוצאות מטבלת הביניים החדשה, ואולי משתמשים בשלב קשור בתהליך העבודה.

בתרשים הבא מוצג התהליך המעודכן שבו טבלת הביניים ב' כוללת את סכימת ב', ומוצג גם איך התצוגה החיצונית מתעדכנת כך שתכלול תוכן מהטבלה הראשית ומשתי טבלאות הביניים.

הצינור משתמש עכשיו בסכימה ב' וכותב לטבלת הביניים ב'. תצוגת חזית קוראת מהטבלה הראשית, מטבלת הביניים א' ומטבלת הביניים ב'.

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

טבלת הביניים A מוזגה לטבלה הראשית. התצוגה החיצונית קוראת מ-Staging Table B ומהטבלה הראשית.

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