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

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

הוספת קוד חיצוני לצינור

ההבדל העיקרי בין צינורות עיבוד מקביליים מאוד לבין צינורות אחרים הוא שהם משתמשים בקוד C++‎ בתוך DoFn ולא באחת משפות ה-SDK הרגילות של Apache Beam. בצינורות Java, כדי להקל על השימוש בספריות C++‎ בצינור, מומלץ להשתמש בהפעלות חיצוניות של פרוצדורות. בקטע הזה מתואר הגישה הכללית שמשמשת להרצת קוד חיצוני (C++) בצינורות Java.

הגדרת צינור עיבוד נתונים של Apache Beam כוללת כמה רכיבים מרכזיים:

  • ‫PCollections הם אוספים שלא ניתן לשנות, של אלמנטים הומוגניים.
  • ‫PTransforms משמשים להגדרת הטרנספורמציות ל-PCollection שיוצר PCollection אחר.
  • הצינור הוא המבנה שמאפשר לכם, באמצעות קוד, להצהיר על האינטראקציות בין PTransforms לבין PCollections. הפייפליין מיוצג כגרף אציקלי מכוון (DAG).

כשמשתמשים בקוד משפה שלא נכללת בשפות הסטנדרטיות של Apache Beam SDK, צריך למקם את הקוד ב-PTransform, שנמצא בתוך DoFn, ולהשתמש באחת משפות ה-SDK הסטנדרטיות כדי להגדיר את צינור הנתונים עצמו. מומלץ להשתמש ב-Apache Beam Python SDK כדי להגדיר את צינור הנתונים, כי ל-Python SDK יש מחלקה של כלי עזר שמפשטת את השימוש בקוד אחר. עם זאת, אפשר להשתמש בערכות SDK אחרות של Apache Beam.

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

בתרשים הבא מוצגים שני סוגי השימוש בנתוני צינור עיבוד הנתונים:

  • הנתונים משמשים לניהול התהליך.
  • הנתונים נאספים במהלך העיבוד ומצורפים לנתוני הנהג.

שני שלבים של נתוני פייפליין

בדף הזה, הנתונים העיקריים (מהמקור) נקראים נתוני נסיעה, והנתונים המשניים (משלב העיבוד) נקראים נתוני הצטרפות.

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

שיקולים לגבי גודל נתוני הנהיגה

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

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

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

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

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

הרצת הקוד החיצוני

יש הרבה דרכים להריץ קוד חיצוני ב-Apache Beam.

  • יוצרים תהליך שמופעל מאובייקט DoFn בתוך טרנספורמציה של Dataflow.

  • משתמשים ב-JNI עם Java SDK.

  • יוצרים תהליך משנה ישירות מאובייקט DoFn. הגישה הזו לא הכי יעילה, אבל היא אמינה ופשוטה להטמעה. בגלל הבעיות הפוטנציאליות בשימוש ב-JNI, בדף הזה מוצג שימוש בקריאה לתהליך משנה.

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

מורידים את הקבצים הבינאריים למארחים

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

קבצים בינאריים שמאוחסנים בקטגוריות

כדי להעביר את הקוד: בדוגמה מוצגים השלבים ל-Apache Beam Java SDK.

  1. אחסון הקוד החיצוני שעבר קומפילציה, יחד עם פרטי ניהול הגרסאות, ב-Cloud Storage.
  2. בשיטה @Setup, יוצרים בלוק מסונכרן כדי לבדוק אם קובץ הקוד זמין במשאב המקומי. במקום להטמיע בדיקה פיזית, אפשר לאשר את הזמינות באמצעות משתנה סטטי כשהשרשור הראשון מסתיים.
  3. אם הקובץ לא זמין, משתמשים בספריית הלקוח של Cloud Storage כדי למשוך את הקובץ מקטגוריה של Cloud Storage אל ה-worker המקומי. הגישה המומלצת היא להשתמש במחלקה Apache Beam FileSystems למשימה הזו.
  4. אחרי העברת הקובץ, מוודאים שביט ההרשאה להרצה מוגדר בקובץ הקוד.
  5. במערכת ייצור, בודקים את הגיבוב של הקבצים הבינאריים כדי לוודא שהקובץ הועתק בצורה נכונה.

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

הפעלת קבצים בינאריים חיצוניים

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

int main(int argc, char* argv[])
{
    if(argc < 3){
        std::cerr << "Required return file and data to process" << '\n';
        return 1;
    }

    std::string returnFile = argv[1];
    std::string word = argv[2];

    std::ofstream myfile;
    myfile.open (returnFile);
    myfile << word;
    myfile.close();
    return 0;
}

הקוד הזה קורא שני פרמטרים מרשימת הארגומנטים. הפרמטר הראשון הוא המיקום של קובץ ההחזרה שאליו הנתונים נדחפים. הפרמטר השני הוא הנתונים שהקוד משקף למשתמש. ביישומים בעולם האמיתי, הקוד הזה יעשה יותר מאשר להדפיס את המחרוזת Hello, world!‎.

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

  1. העברת הנתונים לקובצי הבינאריים של הקוד החיצוני.
  2. מריצים את הקבצים הבינאריים, מאתרים שגיאות ומתעדים את השגיאות והתוצאות.
  3. טיפול בפרטי הרישום.
  4. איסוף נתונים מהעיבוד שהושלם.

העברת הנתונים לקבצים הבינאריים

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

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

  • נתוני קלט קטנים. אם מדובר בנתונים קטנים שלא חורגים מהאורך המקסימלי של ארגומנט של פקודה במערכת, אפשר להעביר את הארגומנט במיקום 2 של התהליך שנוצר באמצעות java.lang.ProcessBuilder.
  • נתוני קלט גדולים. אם הנתונים גדולים יותר, צריך ליצור קובץ שהשם שלו כולל UUID כדי שיכיל את הנתונים שנדרשים לתהליך.

הרצת קוד C++, איתור שגיאות ורישום ביומן

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

הגישה המומלצת היא להפנות את stdout ו-stderr לקבצים, וכך לא צריך להתחשב בבעיות שקשורות לזיכרון. לדוגמה, ב-Dataflow runner שקורא לקוד C++, אפשר לכלול שורות כמו הבאות:

Java 

  import java.lang.ProcessBuilder.Redirect;
  ...
      processbuilder.redirectError(Redirect.appendTo(errfile));
      processbuilder.redirectOutput(Redirect.appendTo(outFile));

Python 

# Requires Apache Beam 2.34 or later.
stopping_times, bad_values = (
    integers
    | beam.Map(collatz.total_stopping_time).with_exception_handling(
        use_subprocess=True))

# Write the bad values to a side channel.
bad_values | 'WriteBadValues' >> beam.io.WriteToText(
    os.path.splitext(output_path)[0] + '-bad.txt')

טיפול במידע מיומני רישום

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

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

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

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

איסוף נתונים מעיבוד שהסתיים

לא מומלץ להשתמש ב-stdout כדי להעביר את תוצאת החישוב בחזרה לפונקציה DoFn. קוד אחר שהקוד שלכם ב-C++ קורא לו, ואפילו הקוד שלכם, עשוי לשלוח הודעות גם אל stdout, ולזהם את הזרם stdoutput שמכיל אחרת נתוני רישום ביומן. במקום זאת, מומלץ לבצע שינוי בקוד העטיפה של C++ כדי לאפשר לקוד לקבל פרמטר שמציין איפה ליצור את הקובץ שבו הערך מאוחסן. מומלץ לאחסן את הקובץ הזה בפורמט ניטרלי לשפה באמצעות protocol buffers, כדי שקוד C++‎ יוכל להעביר אובייקט בחזרה לקוד Java או Python. ‫DoFnobject יכול לקרוא את התוצאה ישירות מהקובץ ולהעביר את פרטי התוצאה לקריאה משלו output.

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

עיצוב עיבוד למחזורי מעבד קטנים

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

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

עם זאת, בתרחישי שימוש אחרים, כמו בתחום הפיננסים, העיבוד דורש יחידות קטנות מאוד של זמן מעבד (עשרות אלפיות שנייה). במקרה כזה, התקורה של קריאה לתהליך המשנה גדולה באופן לא פרופורציונלי. פתרון לבעיה הזו הוא שימוש בטרנספורמציה GroupByKey של Apache Beam כדי ליצור קבוצות של 50 עד 100 רכיבים שיוזנו לתהליך. לדוגמה, אפשר לפעול לפי השלבים הבאים:

  • בפונקציה DoFn, יוצרים צמד מפתח/ערך. אם אתם מעבדים עסקאות פיננסיות, אתם יכולים להשתמש במספר העסקה כמפתח. אם אין לכם מספר ייחודי שאפשר להשתמש בו כמפתח, אתם יכולים ליצור סכום ביקורת מהנתונים ולהשתמש בפונקציית מודולו כדי ליצור מחיצות של 50 רכיבים.
  • שולחים את המפתח לפונקציה GroupByKey.create, שמחזירה אוסף KV<key,Iterable<data>> שמכיל את 50 האלמנטים שאפשר לשלוח לתהליך.

הגבלת מקביליות של worker

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

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

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

שימוש במאגרי נתונים עם קיבולת גבוהה ב- Google Cloud

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

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

מקורות מידע זמינים ב- Google Cloud

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

ניהול שגיאות פילוח

כשמשתמשים בקוד C++ בצינור עיבוד נתונים, צריך להחליט איך לנהל שגיאות פילוח (segfaults), כי אם לא מטפלים בהן בצורה נכונה, הן עלולות להשפיע על חלקים אחרים של הקוד. ה-runner של Dataflow יוצר תהליכים לפי הצורך ב-Java, ב-Python או ב-Go, ואז מקצה עבודה לתהליכים בצורה של חבילות.

אם הקריאה לקוד C++‎ מתבצעת באמצעות כלים עם צימוד הדוק, כמו JNI או Cython, ותהליך C++‎ גורם לשגיאת פילוח (segfault), גם התהליך שקורא למכונה הווירטואלית של Java‏ (JVM) קורס. בתרחיש הזה, אי אפשר לזהות נקודות נתונים בעייתיות. כדי לאתר נקודות נתונים בעייתיות, כדאי להשתמש בצימוד חלש יותר, שמפצל את הנתונים הבעייתיים ומאפשר לצינור עיבוד הנתונים להמשיך. עם זאת, אם יש לכם קוד C++ בוגר שנבדק באופן מלא מול כל וריאציות הנתונים, אתם יכולים להשתמש במנגנונים כמו Cython.

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