קריאה מ-Pub/Sub אל Dataflow

בדף הזה מתוארות שיטות מומלצות לקריאה מ-Pub/Sub ב-Dataflow.

‫Apache Beam מספק הטמעת עזר של מחבר הקלט/פלט של Pub/Sub, לשימוש על ידי רכיבי הרצה שאינם של Dataflow. עם זאת, מריץ המשימות של Dataflow משתמש בהטמעה מותאמת אישית משלו למחבר. ההטמעה הזו מנצלת את ממשקי ה-API והשירותים הפנימיים של Google Cloudכדי להציע סימני מים עם זמן אחזור קצר, דיוק גבוה של סימני מים וביטול כפילויות יעיל לעיבוד הודעות בדיוק פעם אחת. המחבר זמין ל-Java, ל-Python ול-Go.

עיבוד "בדיוק פעם אחת"

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

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

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

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

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

ביטול כפילויות לפי מאפיין הודעה

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

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

למידע נוסף על השימוש במאפייני מזהה, קראו את הנושאים הבאים במאמרי העזרה של ה-SDK:

מינויים

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

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

חותמות זמן וסימני מים

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

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

למידע נוסף על השימוש בחותמות זמן של אירועים, אפשר לעיין בנושאי העזר הבאים של ה-SDK:

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

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

  • pubsub.subscriptions.create
  • pubsub.subscriptions.consume
  • pubsub.subscriptions.delete

בנוסף, נדרשת ההרשאה pubsub.topics.attachSubscription בנושא Pub/Sub. מומלץ ליצור תפקיד בהתאמה אישית בניהול הזהויות והרשאות הגישה (IAM) שמכיל רק את ההרשאות האלה.

מידע נוסף על סימני מים זמין בדף StackOverflow, שמסביר איך Dataflow מחשב סימני מים של Pub/Sub.

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

חיפוש ב-Pub/Sub

Pub/Sub Seek מאפשר למשתמשים להפעיל מחדש הודעות שאושרו בעבר. אתם יכולים להשתמש ב-Pub/Sub Seek עם Dataflow כדי לעבד מחדש הודעות בצינור עיבוד נתונים.

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

כדי לעבד מחדש הודעות באמצעות Pub/Sub Seek, מומלץ להשתמש בתהליך העבודה הבא:

  1. יוצרים תמונת מצב של המינוי.
  2. יוצרים מינוי חדש לנושא Pub/Sub. המינוי החדש יורש את קובץ ה-snapshot.
  3. מרוקנים או מבטלים את משימת Dataflow הנוכחית.
  4. שולחים מחדש את צינור עיבוד הנתונים באמצעות המינוי החדש.

מידע נוסף זמין במאמר עיבוד מחדש של הודעות באמצעות Snapshot ו-Seek ב-Pub/Sub.

מקביליות של מקור Pub/Sub

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

  • במשימות של Streaming Engine, ‏Dataflow מחשב את מספר המפתחות באופן דינמי על סמך ה-workers הפעילים ברגע מסוים: (number of workers * threads per worker) / 2. הערך המחושב מוגבל לטווח שבין 128 ל-65,536 מפתחות, ואז מעוגל כלפי מטה לחזקה הקרובה ביותר של 2 (החזקה הגבוהה ביותר של 2 שקטנה מהערך המחושב או שווה לו).

  • במשימות שלא קשורות למנוע סטרימינג, מספר המפתחות הוא החזקה הנמוכה ביותר של 2 שגדולה מ-(4 * maximum workers) או שווה לו.

כדי לשנות את מספר ברירת המחדל של מפתחות ה-shuffle, מגדירים את num_pubsub_keys אפשרות השירות:

Java 

--dataflowServiceOptions=num_pubsub_keys=NUMBER_OF_KEYS

Python 

--dataflow_service_options=num_pubsub_keys=NUMBER_OF_KEYS

המשך 

--dataflow_service_options=num_pubsub_keys=NUMBER_OF_KEYS

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

לדוגמה, אפשר להגדיר את האפשרות הזו במצבים הבאים:

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

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

תכונות לא נתמכות של Pub/Sub

התכונות הבאות של Pub/Sub לא נתמכות בהטמעה של מחבר הקלט/פלט של Pub/Sub ב-Dataflow runner.

השהיה מעריכית לפני ניסיון חוזר (exponential backoff)

כשיוצרים מינוי ל-Pub/Sub, אפשר להגדיר אותו כך שישתמש במדיניות ניסיון חוזר עם השהיה מעריכית לפני ניסיון חוזר (exponential backoff). עם זאת, השהיה מעריכית לפני ניסיון חוזר (exponential backoff) לא פועלת עם Dataflow. במקום זאת, צריך ליצור את המינוי עם מדיניות הניסיון החוזר Retry immediately.

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

נושאים להודעות ללא מוצא

אל תשתמשו בנושאים להודעות ללא מוצא ב-Pub/Sub עם Dataflow, מהסיבות הבאות:

  • ‫Dataflow שולח אישורים שליליים מסיבות פנימיות שונות (לדוגמה, אם ה-worker מושבת). כתוצאה מכך, יכול להיות שהודעות יועברו לנושא להודעות ללא מוצא גם אם לא מתרחשות שגיאות בקוד של צינור עיבוד הנתונים.

  • ‫Dataflow מאשר הודעות אחרי שחבילת הודעות מעובדת בהצלחה בשלב הממוזג הראשון. אם לצינור עיבוד הנתונים יש כמה שלבים משולבים, והכשלים מתרחשים בכל שלב אחרי השלב הראשון, ההודעות כבר אושרו ולא עוברות לנושא להודעות ללא מוצא.

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

מסירה פעם אחת בדיוק ב-Pub/Sub

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

סדר ההודעות ב-Pub/Sub

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

לא מומלץ להשתמש בסדר ההודעות ב-Dataflow, מהסיבות הבאות:

  • יכול להיות שהמחבר של Pub/Sub I/O לא ישמור על סדר ההודעות.
  • ב-Apache Beam אין הנחיות מחמירות לגבי הסדר שבו הרכיבים מעובדים. לכן, יכול להיות שהסדר לא יישמר בטרנספורמציות במורד הזרם.
  • שימוש בסדר ההודעות של Pub/Sub עם Dataflow עלול להגדיל את זמן האחזור ולהפחית את הביצועים.

טרנספורמציות של הודעה אחת ב-Pub/Sub

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

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