עבודה עם צינורות נתונים של Dataflow

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

הערה: אפשר לדווח על בעיות בצינורות נתונים של Dataflow ולבקש תכונות חדשות בכתובת google-data-pipelines-feedback.

סקירה כללית

Dataflow Data Pipelines הם שירות מנוהל במלואו להמרת נתונים ולהוספת נתונים במצבי סטרימינג ואצווה. בעזרת Dataflow Data Pipelines תוכלו:

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

מידע נוסף על ה-API מופיע במאמרי העזרה של ה-API בנושא Data Pipelines.

תכונות

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

מגבלות

• זמינות אזורית: אפשר ליצור פייפליינים של נתונים באזורים זמינים של Cloud Scheduler.

• מכסה:

  • מספר ברירת המחדל של צינורות עיבוד נתונים לכל פרויקט: 500

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

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

  הערה: כדי להגדיל את המכסה, צריך לפנות Google Cloud לתמיכה. עם זאת, ערך המגבלה כפוף למכסות ולמגבלות של Cloud Scheduler ושל Dataflow.

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

סוגים של פייפליינים

ב-Dataflow יש שני סוגים של פייפליינים: סטרימינג ואצווה. שני הסוגים של צינורות עיבוד נתונים מריצים משימות שמוגדרות בתבניות של Dataflow.

פייפליין של נתונים בסטרימינג

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

פייפליין של נתונים באצווה

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

הערה: כל שם של עבודת אצווה ב-Dataflow שנוצר על ידי פייפליין של נתוני אצווה משתמש בתבנית השמות הבאה: <pipeline_id>-mp--<timestamp>-<random int>. הערך של timestamp הוא ברמת דיוק של שניות. המחרוזת -mp– שמורה לציון העובדה שמשימת אצווה של Dataflow נוצרה על ידי פייפליין של נתונים באצווה. הטקסט pipeline_id נחתך אחרי 27 תווים.

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

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

• אפשר להשתמש במחזיקי מקום לשנה, לחודש, לתאריך, לשעה, לדקה ולשנייה, והם צריכים להיות בפורמט strftime(). לפני מחזיקי המקום מופיע סמל האחוז (%).
• הפורמט של הפרמטר לא מאומת במהלך יצירת צינור הנתונים.
  • דוגמה: אם מציינים את הנתיב המפורמט לקלט gs://bucket/Y, הוא מוערך כ-gs://bucket/Y, כי Y ללא % לפניו לא ממופה לפורמט strftime().

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

• דוגמה: צינור להעברת נתונים באצווה מתוזמן לחזור על עצמו בתחילת כל שעה לפי שעון החוף הפסיפי (PST). אם מגדירים את נתיב הקלט כפרמטר gs://bucket-name/%Y-%m-%d/prefix-%H_%M.csv, ב-15 באפריל 2021 בשעה 18:00 (לפי שעון החוף המערבי), נתיב הקלט יהיה gs://bucket-name/2021-04-15/prefix-18_00.csv.

שימוש בפרמטרים של הזזת זמן

אפשר להשתמש בפרמטרים של הזזת זמן של דקה או שעה עם סימן + או -. כדי לתמוך בהתאמה של נתיב קלט עם תאריך ושעה מוערכים שמוזזים לפני או אחרי התאריך והשעה הנוכחיים של לוח הזמנים של צינור הנתונים, צריך להוסיף סוגריים מסולסלים לפרמטרים האלה. בפורמט {[+|-][0-9]+[m|h]}. צינור הנתונים של העיבוד באצווה ממשיך לחזור על עצמו בזמן המתוזמן שלו, אבל נתיב הקלט מוערך עם היסט הזמן שצוין.

• דוגמה: צינור להעברת נתונים באצווה מתוזמן לחזור על עצמו בתחילת כל שעה לפי שעון החוף הפסיפי (PST). אם מגדירים את נתיב הקלט כפרמטר gs://bucket-name/%Y-%m-%d/prefix-%H_%M.csv{-2h}, ב-15 באפריל 2021 בשעה 18:00 (לפי שעון החוף המערבי), נתיב הקלט יהיה gs://bucket-name/2021-04-15/prefix-16_00.csv.

תפקידים בפייפליין

כדי שהפעולות של פייפליין הנתונים של Dataflow יתבצעו בהצלחה, אתם צריכים את תפקידי ה-IAM הנדרשים, כמו שמתואר בהמשך:

1. כדי לבצע פעולות, צריך את התפקיד המתאים:

   • ‫Datapipelines.admin: יכול לבצע את כל הפעולות בפייפליין של הנתונים
   • ‫Datapipelines.viewer: יכולים לצפות בצינורות עיבוד נתונים ובעבודות
   • ‫Datapipelines.invoker: יכול להפעיל ריצה של משימת פייפליין נתונים (אפשר להפעיל את התפקיד הזה באמצעות ה-API)

2. לחשבון השירות שבו משתמש Cloud Scheduler צריך להיות התפקיד roles/iam.serviceAccountUser, בין אם חשבון השירות מוגדר על ידי המשתמש או שהוא חשבון השירות שמוגדר כברירת מחדל ב-Compute Engine. מידע נוסף זמין במאמר בנושא תפקידים ב-Data pipelines.

3. צריך לקבל את התפקיד roles/iam.serviceAccountUser בחשבון השירות שבו משתמשים ב-Cloud Scheduler וב-Dataflow, כדי שתוכלו לפעול בשם חשבון השירות הזה. אם לא בוחרים חשבון שירות ל-Cloud Scheduler ול-Dataflow, נעשה שימוש בחשבון השירות שמוגדר כברירת מחדל ב-Compute Engine.

הערה: אם נתקלתם בסטטוס השגיאה NOT_FOUND מ-Cloud Scheduler עם רשומה ביומן type.googleapis.com/google.cloud.scheduler.logging.AttemptFinished, זה בדרך כלל מצביע על הרשאות חסרות בחשבונות השירות של Cloud Scheduler או של Dataflow. כדי לפתור בעיות בפייפליין, צריך לאמת ולתקן את ההרשאות האלה.

יצירת פייפליין

אפשר ליצור פייפליין נתונים של Dataflow בשתי דרכים:

1. ייבוא משרה, או
2. יצירת פייפליין

דף ההגדרה של Data pipelines: כשניגשים בפעם הראשונה לתכונה של Dataflow ב- Google Cloud Console, נפתח דף הגדרה. מפעילים את ממשקי ה-API שמופיעים ברשימה כדי ליצור פייפליינים.

ייבוא משרה

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

1. נכנסים לדף Jobs של Dataflow במסוף Google Cloud .

   מעבר לדף Jobs

2. בוחרים משימה שהושלמה, ואז בדף פרטי המשימה לוחצים על +ייבוא כצינור.

3. בדף Create pipeline from template (יצירת צינור נתונים מתבנית), הפרמטרים מאוכלסים באפשרויות של המשימה שיובאה.

4. עבור משימה באצווה, בקטע Schedule your pipeline (תזמון צינור הנתונים), מציינים לוח זמנים של חזרה. לא חובה לספק כתובת של חשבון אימייל ל-Cloud Scheduler, שמשמש לתזמון של הפעלות אצווה. אם לא מציינים חשבון שירות, המערכת משתמשת בחשבון השירות שמוגדר כברירת מחדל ב-Compute Engine.

יצירת פייפליין

1. נכנסים לדף Data pipelines של Dataflow במסוף Google Cloud .

   מעבר אל 'צינורות נתונים'

2. לוחצים על +יצירת פייפליין.

3. בדף Create pipeline from template (יצירת צינור עיבוד נתונים מתבנית), מציינים שם לצינור וממלאים את שאר השדות של בחירת התבנית והפרמטרים.

4. עבור משימה באצווה, בקטע Schedule your pipeline (תזמון צינור הנתונים), צריך לציין תזמון חוזר. לא חובה לספק כתובת של חשבון אימייל ל-Cloud Scheduler, שמשמש לתזמון של הפעלות אצווה. אם לא מציינים ערך, נעשה שימוש בחשבון השירות המוגדר כברירת מחדל ב-Compute Engine.

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

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

• קטגוריה של Cloud Storage לאחסון קובצי קלט ופלט
• מערך נתונים ב-BigQuery שבו תיצור טבלה.

בצינור הזה לדוגמה נעשה שימוש בתבנית של צינור אצווה Cloud Storage Text to BigQuery. התבנית הזו קוראת קבצים בפורמט CSV מ-Cloud Storage, מריצה טרנספורמציה ואז מוסיפה ערכים לטבלה ב-BigQuery עם שלוש עמודות.

1. יוצרים את הקבצים הבאים בכונן המקומי:

   1. קובץ bq_three_column_table.json שמכיל את הסכימה הבאה של טבלת היעד ב-BigQuery.

      {
        "BigQuery Schema": [
          {
            "name": "col1",
            "type": "STRING"
          },
          {
            "name": "col2",
            "type": "STRING"
          },
          {
            "name": "col3",
            "type": "INT64"
          }
        ]
      }
      

   2. קובץ split_csv_3cols.js JavaScript‏, שמבצע טרנספורמציה בסיסית בנתוני הקלט לפני ההוספה ל- BigQuery.

      function transform(line) {
          var values = line.split(',');
          var obj = new Object();
          obj.col1 = values[0];
          obj.col2 = values[1];
          obj.col3 = values[2];
          var jsonString = JSON.stringify(obj);
          return jsonString;
      }
      

   3. קובץ CSV‏ file01.csv עם כמה רשומות שמוכנסות לטבלה ב-BigQuery.

      b8e5087a,74,27531
      7a52c051,4a,25846
      672de80f,cd,76981
      111b92bf,2e,104653
      ff658424,f0,149364
      e6c17c75,84,38840
      833f5a69,8f,76892
      d8c833ff,7d,201386
      7d3da7fb,d5,81919
      3836d29b,70,181524
      ca66e6e5,d7,172076
      c8475eb6,03,247282
      558294df,f3,155392
      737b82a8,c7,235523
      82c8f5dc,35,468039
      57ab17f9,5e,480350
      cbcdaf84,bd,354127
      52b55391,eb,423078
      825b8863,62,88160
      26f16d4f,fd,397783
      

2. כדי להעתיק את הקבצים לתיקיות בקטגוריה של Cloud Storage בפרויקט, משתמשים בפקודה gcloud storage cp באופן הבא:

   1. העתקת bq_three_column_table.json ו-split_csv_3cols.js אל gs://BUCKET_ID/text_to_bigquery/

      gcloud storage cp bq_three_column_table.json gs://BUCKET_ID/text_to_bigquery/
      gcloud storage cp split_csv_3cols.js gs://BUCKET_ID/text_to_bigquery/

   2. העתקה של file01.csv אל gs://BUCKET_ID/inputs/

      gcloud storage cp file01.csv gs://BUCKET_ID/inputs/

3. במסוף Google Cloud , נכנסים לדף Buckets של Cloud Storage.

   כניסה לדף Buckets

4. כדי ליצור תיקייה tmp בקטגוריה של Cloud Storage, בוחרים את שם התיקייה כדי לפתוח את דף הפרטים של הקטגוריה, ואז לוחצים על יצירת תיקייה.

   

5. נכנסים לדף Data pipelines של Dataflow במסוף Google Cloud .

   מעבר אל 'צינורות נתונים'

6. בוחרים באפשרות יצירת פייפליין. בדף יצירת צינור מכירות מתבנית, מזינים או בוחרים את הפריטים הבאים:

   1. בשדה Pipeline name (שם צינור), מזינים text_to_bq_batch_data_pipeline.
   2. בקטע Regional endpoint (נקודת קצה אזורית), בוחרים אזור ב-Compute Engine. אזורי המקור והיעד צריכים להיות זהים. לכן, הקטגוריה של Cloud Storage והטבלה ב-BigQuery צריכים להיות באותו אזור.

   3. בקטע תבנית Dataflow, בעיבוד נתונים בכמות גדולה (באצווה), בוחרים באפשרות קבצי טקסט ב-Cloud Storage ל-BigQuery.

      הערה: אל תבחרו את צינור הנתונים להזרמה עם אותו שם באפשרות עיבוד נתונים באופן רציף (הזרמה).

   4. בקטע Schedule your pipeline (תזמון צינור הנתונים), בוחרים תזמון, למשל Hourly (שעתי) בדקה 25 באזור הזמן שלכם. אפשר לערוך את לוח הזמנים אחרי ששולחים את צינור הנתונים. אפשר לספק כתובת אימייל לחשבון Cloud Scheduler, שמשמש לתזמון של הפעלות אצווה. אם לא מציינים חשבון שירות, המערכת משתמשת בחשבון השירות שמוגדר כברירת מחדל ב-Compute Engine.

   5. בקטע פרמטרים נדרשים, מזינים את הפרטים הבאים:

      1. נתיב של פונקציית UDF ב-JavaScript ב-Cloud Storage:

         gs://BUCKET_ID/text_to_bigquery/split_csv_3cols.js
         

      2. בקטע נתיב JSON:

         BUCKET_ID/text_to_bigquery/bq_three_column_table.json
         

      3. בשדה JavaScript UDF name (שם פונקציית UDF של JavaScript): transform
      4. טבלת פלט של BigQuery:

         PROJECT_ID:DATASET_ID.three_column_table
         

      5. בקטע Cloud Storage input path (נתיב קלט של Cloud Storage):

         BUCKET_ID/inputs/file01.csv
         

      6. בקטע ספריית BigQuery זמנית:

         BUCKET_ID/tmp
         

      7. במיקום זמני:

         BUCKET_ID/tmp
         

   6. לוחצים על יצירת צינור.

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

   

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

אפשר גם להפעיל צינור עיבוד נתונים באצווה לפי דרישה באמצעות הלחצן הפעלה במסוף Dataflow Pipelines.

יצירת פייפליין נתונים לדוגמה בסטרימינג

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

• בקטע Pipeline schedule (תזמון של צינור עיבוד הנתונים), לא מציינים תזמון לצינור נתונים בסטרימינג. המערכת תתחיל מיד את משימת הסטרימינג של Dataflow.
• בקטע תבנית Dataflow, באפשרות עיבוד נתונים באופן רציף (סטרימינג), בוחרים באפשרות קבצי טקסט ב-Cloud Storage ל-BigQuery.
• ב-Worker machine type, הפייפליין מעבד את קבוצת הקבצים הראשונית שתואמת לתבנית gs://BUCKET_ID/inputs/file01.csv ואת כל הקבצים הנוספים שתואמים לתבנית הזו שמעלים לתיקייה inputs/. אם הגודל של קובצי ה-CSV גדול מכמה GB, כדי להימנע משגיאות אפשריות של חוסר זיכרון, צריך לבחור סוג מכונה עם זיכרון גדול יותר מסוג המכונה שמוגדר כברירת מחדל עם 4 vCPU, כמו מכונה מרובת זיכרון (high-memory) עם 8 vCPU (לדוגמה, n4-highmem-8).

פתרון בעיות

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

הפעלת משימה של פייפליין נתונים נכשלת

כשמשתמשים ב-Data Pipelines כדי ליצור תזמון של משימה חוזרת, יכול להיות שהמשימה ב-Dataflow לא תופעל, ושגיאת סטטוס 503 תופיע בקובצי היומן של Cloud Scheduler.

הבעיה הזו מתרחשת כש-Dataflow לא מצליח להריץ את העבודה באופן זמני.

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

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

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

צינורות עיבוד נתונים חוזרים של קבוצות

כדי לבצע ניתוח ראשוני של תקינות צינור עיבוד הנתונים, משתמשים בתרשימים Individual job status (סטטוס של כל משימה) ו-Thread time per step (זמן השרשור לכל שלב) בדף Pipeline info (פרטי צינור עיבוד הנתונים) במסוף Google Cloud . התרשימים האלה נמצאים בחלונית 'סטטוס צינורות'.

דוגמה לחקירה:

1. יש לכם צינור להרצת משימות באצווה שחוזר על עצמו כל שעה, 3 דקות אחרי תחילת השעה. כל משימה פועלת בדרך כלל במשך כ-9 דקות. יש לכם מטרה לסיים את כל העבודות בפחות מ-10 דקות.

2. בתרשים של סטטוס העבודה מוצג שעבודה מסוימת רצה במשך יותר מ-10 דקות.

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

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

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

דוגמה לחקירה:

1. יש לכם פייפליין זורם שבדרך כלל יוצר פלט עם עדכניות הנתונים של 20 שניות.

2. הגדרתם יעד של עדכניות נתונים למשך 30 שניות. כשבודקים את הגרף של עדכניות הנתונים, אפשר לראות שבין 9:00 ל-10:00, עדכניות הנתונים עלתה כמעט ל-40 שניות.

   

3. עוברים לכרטיסייה Pipeline metrics וצופים בתרשימים CPU Utilization ו-Memory Utilization כדי לבצע ניתוח נוסף.

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

אם מנסים ליצור צינור חדש עם שם שכבר קיים בפרויקט, מוצגת הודעת השגיאה Pipeline Id already exist within the project. כדי להימנע מהבעיה הזו, חשוב לבחור תמיד שמות ייחודיים לצינורות.

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