יצירת משימה בהתאמה אישית באמצעות כלי ליצירת משימות

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

יצירת פייפליין חדש

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

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

   מעבר אל Jobs

2. לוחצים על Create job from builder (יצירת משימה מכלי הבנייה).

3. בשדה Job name, מזינים שם למשימה.

4. בוחרים באפשרות Batch או Streaming.

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

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

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

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

הוספת מקור לצינור

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

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

2. ברשימה סוג המקור, בוחרים את סוג מקור הנתונים.

3. בהתאם לסוג המקור, מספקים פרטי הגדרה נוספים.

   • לדוגמה, אם בוחרים ב-BigQuery, צריך לציין את הטבלה שממנה רוצים לקרוא.
   • אם בוחרים באפשרות Pub/Sub, צריך לציין סכימת הודעות. מזינים את השם ואת סוג הנתונים של כל שדה שרוצים לקרוא מהודעות Pub/Sub. הצינור משמיט את כל השדות שלא צוינו בסכימה.
   • אם בוחרים באפשרות Apache Iceberg, צריך לציין את פרטי החיבור לקטלוג REST של Iceberg‏ (IRC), כמו מזהה טבלת Iceberg, שם הקטלוג, סוג הקטלוג, URI של הקטלוג ושם המחסן.

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

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

הוספת טרנספורמציה לצינור

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

סוג הטרנספורמציה	תיאור	מידע על Beam YAML Transform
מסנן (Python)	סינון רשומות באמצעות ביטוי Python.	סינון (Python) סינון
SQL Transform	אפשר לשנות רשומות או לצרף כמה קלטים באמצעות הצהרת SQL.	טרנספורמציות עם שם Sql
מיפוי שדות (Python)	להוסיף שדות חדשים או למפות מחדש רשומות שלמות באמצעות ביטויים ופונקציות של Python.	פונקציות מיפוי MapToFields
מיפוי שדות (SQL)	הוספה או מיפוי של שדות רשומה באמצעות ביטויי SQL.	פונקציות מיפוי MapToFields
‫YAML Transforms: AssertEqual AssignTimestamps שילוב Explode מסנן לשטח הצטרפות LogForTesting MLTransform MapToFields PyTransform WindowInfo	משתמשים בכל טרנספורמציה מ-Beam YAML SDK. הגדרת טרנספורמציה של YAML: מספקים את פרמטרי ההגדרה של טרנספורמציית ה-YAML כמפת YAML. הצמדים של מפתח וערך משמשים לאכלוס קטע ההגדרה של טרנספורמציית ה-YAML של Beam שמתקבלת. לפרמטרים של ההגדרות הנתמכות לכל סוג של טרנספורמציה, אפשר לעיין במסמכי התיעוד של טרנספורמציות Beam YAML. פרמטרים לדוגמה להגדרה: שילוב group_by: combine: הצטרפות type: equalities: fields:	Beam YAML Transform Index
יומן	רישום יומנים ביומני העובדים של המשרה.	LogForTesting
קיבוץ לפי	אפשר לשלב רשומות עם פונקציות כמו count() ו-sum().	Group By
הצטרפות	שילוב של כמה קלטים בשדות שווים.	Beam YAML Join הצטרפות
Explode	פיצול רשומות על ידי ביטול הקיבוץ של שדות מערך.	Flatmap פירוק

כדי להוסיף המרה:

1. לוחצים על הוספת טרנספורמציה.

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

3. ברשימה סוג הטרנספורמציה, בוחרים את סוג הטרנספורמציה.

4. בהתאם לסוג השינוי, מספקים פרטי הגדרה נוספים. לדוגמה, אם בוחרים באפשרות Filter (Python), מזינים ביטוי Python לשימוש כמסנן.

5. בוחרים את שלב הקלט של הטרנספורמציה. שלב הקלט הוא המקור או הטרנספורמציה שהפלט שלהם מספק את הקלט לטרנספורמציה הזו.

   הערה: לטרנספורמציות SQL ו-Join יכולים להיות כמה שלבי קלט.

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

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

1. בתיבה Sink name (שם היעד), מזינים שם ליעד או משתמשים בשם ברירת המחדל. השם מופיע בתרשים העבודה כשמריצים את העבודה.

2. ברשימה Sink type (סוג היעד), בוחרים את סוג היעד.

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

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

5. כדי להוסיף עוד יעד לצנרת, לוחצים על הוספת יעד.

הרצת צינור עיבוד הנתונים

כדי להריץ צינור מ-Job Builder, מבצעים את השלבים הבאים:

1. אופציונלי: הגדרת אפשרויות למשימת Dataflow. כדי להרחיב את הקטע Dataflow options (אפשרויות של Dataflow), לוחצים על החץ להרחבה .

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

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

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

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

1. לוחצים על אימות כדי לפתוח את Cloud Shell ולהתחיל את שירות האימות.
2. לוחצים על התחלת האימות.
3. אם נמצאה שגיאה במהלך האימות, יופיע סימן קריאה אדום.
4. מתקנים את כל השגיאות שאותרו ומאמתים את התיקונים באמצעות לחיצה על אימות. אם לא נמצאה שגיאה, יופיע סימן וי ירוק.

הרצה באמצעות ה-CLI של gcloud

אפשר גם להריץ צינורות Beam YAML באמצעות ה-CLI של gcloud. כדי להריץ צינור עיבוד נתונים של Job Builder באמצעות ה-CLI של gcloud:

1. לוחצים על שמירת YAML כדי לפתוח את החלון שמירת YAML.

2. מבצעים אחת מהפעולות הבאות:

   • כדי לשמור ב-Cloud Storage, מזינים נתיב ב-Cloud Storage ולוחצים על שמירה.
   • כדי להוריד קובץ מקומי, לוחצים על הורדה.

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

     gcloud dataflow yaml run my-job-builder-job --yaml-pipeline-file=YAML_FILE_PATH
   

   מחליפים את YAML_FILE_PATH בנתיב של קובץ ה-YAML, באופן מקומי או ב-Cloud Storage.

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

• שימוש בממשק למעקב אחרי משימות ב-Dataflow.
• שמירה וטעינה של הגדרות משימות ב-YAML בכלי ליצירת משימות.
• מידע נוסף על Beam YAML