תזמון הרצות

במסמך הזה מוסבר איך לבצע את הפעולות הבאות ב-Dataform:

• תזמון הרצות באמצעות הגדרות של תהליכי עבודה.
• תזמון הרצות באמצעות Workflows ו-Cloud Scheduler.
• תזמון הרצות באמצעות Managed Service for Apache Airflow.
• אוטומציה של הרצות באמצעות טריגרים לפיתוח גרסת Build של Cloud Build.

בטבלה הבאה מוצגת השוואה בין השיטות:

‏Method	מטרה	סוג הטריגר
הגדרות של תהליכי עבודה	הרצות ייצור סטנדרטיות מבוססות-זמן ב-Dataform.	מבוסס זמן
Workflows ו-Cloud Scheduler	תזמור קל משקל בלי שרת (serverless) של Dataform לצד ממשקי API אחרים.	מבוסס-זמן או מבוסס-אירוע
Managed Airflow	צינורות עיבוד נתונים מורכבים עם יחסי תלות מחוץ ל-BigQuery.	מבוסס-זמן או מבוסס-אירוע
טריגרים של Cloud Build	אוטומציה של גרסאות מיד אחרי דחיפה של Git.	מבוסס-אירועים

לפני שמתחילים

כדי לתזמן הפעלות באמצעות הגדרות של תהליכי עבודה או לתזמן הפעלות באמצעות תהליכי עבודה ו-Cloud Scheduler, מבצעים את הפעולות הבאות:

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

   כניסה ל-Dataform

2. בוחרים מאגר או יוצרים מאגר.

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

3. יוצרים הגדרת גרסה.

כדי לתזמן הרצות באמצעות Managed Service for Apache Airflow:

1. בוחרים או יוצרים מאגר ב-Dataform.
2. נותנים ל-Dataform גישה ל-BigQuery.
3. בוחרים או יוצרים סביבת עבודה ב-Dataform.
4. יוצרים לפחות טבלה אחת.
5. יוצרים סביבת Managed Airflow 2.

התפקידים הנדרשים

כדי לקבל את ההרשאות שדרושות לביצוע המשימות שמפורטות במאמר הזה, בקשו מהאדמין להקצות לכם את התפקידים הבאים ב-IAM:

• אדמין ב-Dataform (roles/dataform.admin) במאגרים
• משתמש בחשבון שירות (roles/iam.serviceAccountUser) בחשבון השירות בהתאמה אישית
• Composer Worker (roles/composer.worker) בחשבון השירות של סביבת Managed Airflow
• הפעלה אוטומטית באמצעות Cloud Build:
  • אדמין של חשבון שירות (roles/iam.serviceAccountAdmin) בחשבון השירות בהתאמה אישית
  • עריכה ב-Cloud Build (roles/cloudbuild.builds.editor) בפרויקט

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

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

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

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

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

כדי להפעיל הפעלות מתוזמנות של הגדרת תהליך עבודה, צריך לתת את ההרשאה iam.serviceAccounts.actAs לסוכן השירות שמשמש כברירת מחדל של Dataform בחשבון השירות בהתאמה אישית שמשמש בהגדרת תהליך העבודה. ההרשאה הזו זמינה בתפקיד "משתמש בחשבון שירות" (roles/iam.serviceAccountUser). מידע נוסף זמין במאמר שימוש במצב act-as מחמיר.

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

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

תזמון הפעלות באמצעות הגדרות של תהליכי עבודה

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

מידע על הגדרות של תהליכי עבודה

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

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

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

הגדרת תהליך עבודה ב-Dataform כוללת את הגדרות ההרצה הבאות:

• המזהה של הגדרת תהליך העבודה.
• הגדרת הגרסה.

• חשבון שירות.

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

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

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

• פרטי הכניסה של המשתמש לחשבון Google

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

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

• פעולות בתהליך העבודה שצריך לבצע:

  • כל הפעולות.
  • מבחר פעולות.
  • מבחר תגים.

• לוח הזמנים להפעלה ואזור הזמן.

יצירת הגדרה של תהליך עבודה

כדי ליצור הגדרת תהליך עבודה ב-Dataform:

1. במאגר, עוברים אל גרסאות ותזמון.
2. בקטע Workflow configurations, לוחצים על Create.

3. בחלונית Create workflow configuration, בשדה Configuration ID, מזינים מזהה ייחודי להגדרת תהליך העבודה.

   המזהים יכולים לכלול רק מספרים, אותיות, מקפים וקווים תחתונים.

4. בתפריט Release configuration, בוחרים הגדרת גרסה של הידור.

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

   • כדי להשתמש בפרטי הכניסה של המשתמש בחשבון Google, בוחרים באפשרות Execute with my user credentials.

     אופציונלי: בקטע Extended access options, בוחרים את השירותים הנוספים שנדרשים לתהליך העבודה:

     • ‫Knowledge Catalog: מאפשר עדכוני מטא-נתונים של Google Cloud Knowledge Catalog.
     • ‫Google Drive: מאפשר גישה לקריאה בלבד לקבצים ב-Google Drive.
     • ‫Bigtable: הרשאת קריאה בלבד לנתונים ב-Google Bigtable.

   • כדי להשתמש בחשבון שירות בהתאמה אישית, בוחרים באפשרות Execute with selected service account ואז בוחרים את חשבון השירות שמשויך לפרויקט ב- Google Cloud שיש לכם גישה אליו. אם לא בוחרים חשבון שירות, הגדרת תהליך העבודה משתמשת בחשבון השירות של המאגר.

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

6. אופציונלי: בשדה Schedule frequency, מזינים את תדירות ההרצות בפורמט unix-cron.

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

7. אופציונלי: בתפריט Timezone, בוחרים את אזור הזמן של ההרצות.

   אזור הזמן שמוגדר כברירת מחדל הוא UTC.

8. בוחרים את הפעולות בתהליך העבודה שרוצים להפעיל:

   • כדי להפעיל את כל תהליך העבודה, לוחצים על כל הפעולות.
   • כדי לבצע פעולות שנבחרו בתהליך העבודה, לוחצים על Selection of actions ובוחרים פעולות.
   • כדי לבצע פעולות עם תגים שנבחרו, לוחצים על Selection of tags ובוחרים תגים.
   • אופציונלי: כדי להפעיל פעולות או תגים נבחרים ואת יחסי התלות שלהם, בוחרים באפשרות Include dependencies.
   • אופציונלי: כדי להפעיל פעולות או תגים נבחרים ואת התלויות שלהם, בוחרים באפשרות Include dependents.

   • אופציונלי: כדי לבנות מחדש את כל הטבלאות, בוחרים באפשרות Run with full refresh.

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

   • אופציונלי: מגדירים את העדיפות של משימת השאילתה ב-BigQuery באמצעות האפשרות Execute as interactive job with high priority (default). כברירת מחדל, BigQuery מריץ שאילתות כמשימות של שאילתות אינטראקטיביות, שנועדו להתחיל לפעול במהירות האפשרית. אם לא בוחרים באפשרות הזו, השאילתות מופעלות כמשימות של שאילתות באצווה, שהעדיפות שלהן נמוכה יותר.

9. לוחצים על יצירה. אם בחרתם באפשרות Execute with my user credentials בתור שיטת האימות, אתם צריכים לאשר את חשבון Google.

לדוגמה, הגדרת תהליך העבודה הבאה מריצה פעולות עם התג hourly בכל שעה באזור הזמן CEST:

• מזהה ההגדרה: production-hourly
• הגדרת הגרסה: -
• תדירות: 0 * * * *
• אזור זמן: Central European Summer Time (CEST)
• בחירה של פעולות בתהליך העבודה: בחירה של תגים, תג hourly

אישור של חשבון Google

כדי לאמת את המשאב באמצעות פרטי הכניסה של חשבון Google, צריך לתת באופן ידני הרשאה לצינורות עיבוד הנתונים של BigQuery לקבל את אסימון הגישה לחשבון Google ולגשת לנתוני המקור בשמכם. אתם יכולים לאשר ידנית באמצעות ממשק תיבת הדו-שיח של OAuth. אם בוחרים באחת מאפשרויות הגישה המורחבת, צריך לתת גישה לשירותים האלה, למשל Google Drive או Knowledge Catalog.

הערה: כללי מדיניות של בקרת גישה מבוססת-הקשר (CAA) – כולל כללי מדיניות שמבוססים על כתובת IP, על מיקום גיאוגרפי ועל תאימות מכשירים – לא נתמכים כשמפעילים או מתזמנים צינורות עיבוד נתונים של BigQuery עם פרטי כניסה של משתמש לחשבון Google, כי הבקשות לאסימונים מגיעות מהתשתית של Google. כללי המדיניות של CAA חוסמים את ההפעלות האלה, אלא אם מזהה הלקוח של Dataform ב-OAuth פטור מכללי המדיניות.

צריך לתת הרשאה לצינורות עיבוד הנתונים של BigQuery רק פעם אחת.

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

1. עוברים אל הדף של החשבון ב-Google.
2. לוחצים על BigQuery Pipelines.
3. לוחצים על הסרת הגישה.

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

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

עריכת הגדרה של תהליך עבודה

כדי לערוך הגדרה של תהליך עבודה, פועלים לפי השלבים הבאים:

1. במאגר, עוברים אל גרסאות ותזמון.
2. בתפריט More, לוחצים על Edit ליד הגדרת תהליך העבודה שרוצים לערוך.
3. בחלונית Edit workflow configuration, עורכים את הגדרות תצורת תהליך העבודה ולוחצים על Save.

מחיקת הגדרה של תהליך עבודה

כדי למחוק הגדרה של תהליך עבודה:

1. במאגר, עוברים אל גרסאות ותזמון.
2. בתפריט More של הגדרת תהליך העבודה שרוצים למחוק, לוחצים על Delete.
3. בתיבת הדו-שיח Delete release configuration, לוחצים על Delete.

תזמון הרצות באמצעות Workflows ו-Cloud Scheduler

בקטע הזה מוסבר איך לתזמן הפעלות של תהליכי עבודה ב-Dataform באמצעות Workflows ו-Cloud Scheduler.

מידע על הרצות מתוזמנות של תהליכי עבודה

כדי להגדיר את תדירות ההרצה של תהליכי העבודה ב-Dataform, אתם יכולים ליצור משימה ב-Cloud Scheduler שתפעיל תהליך עבודה ב-Workflows. הכלי Workflows מפעיל שירותים בתהליך עבודה של תזמור שאתם מגדירים.

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

הערה: כשמתזמנים הפעלות באמצעות Workflows, כדי להפעיל את createCompilationResult עם ערך gitCommitish, כמו הסתעפות, תג או SHA של commit, צריך לשכפל את מאגר Git המרוחק של Dataform דרך הרשת בכל הפעלה. חיבורי רשת חיצוניים יכולים להיות לא אמינים באופן זמני במהלך הידורים תכופים או בו-זמניים, לכן מומלץ לתזמן תהליכי עבודה בסביבת הייצור באמצעות הגדרות גרסה, או להגדיר ניסיונות חוזרים אוטומטיים עם השהיה מעריכית לפני ניסיון חוזר (exponential backoff) בתהליך העבודה של התזמור. מידע נוסף זמין במאמר לא ניתן לגשת למאגר המרוחק.

יצירת תהליך עבודה מתוזמן של תזמור

כדי לתזמן הרצות של תהליך העבודה ב-Dataform, משתמשים ב-Workflows כדי ליצור תהליך עבודה של תזמור ומוסיפים משימה של Cloud Scheduler כטריגר.

1. ב-Workflows נעשה שימוש בחשבונות שירות כדי לתת לתהליכי עבודה גישה למשאביGoogle Cloud . יוצרים חשבון שירות ונותנים לו את ההרשאות הבאות:

   • תפקיד עריכה ב-Dataform (roles/dataform.editor).
   • התפקיד "משתמש בחשבון שירות" (roles/iam.serviceAccountUser) בחשבון השירות בהתאמה אישית שמשמש ב-Dataform.
   • ההרשאות המינימליות שנדרשות כדי לנהל את תהליך העבודה של התזמור. מידע נוסף זמין במאמר מתן הרשאה לתהליך עבודה כדי לגשת למשאבי Google Cloud .

2. יוצרים תהליך עבודה של תזמור ומשתמשים בקוד המקור הבא ב-YAML בתור ההגדרה של תהליך העבודה:

   main:
       steps:
       - init:
           assign:
           - repository: projects/PROJECT_ID/locations/REPOSITORY_LOCATION/repositories/REPOSITORY_ID
       - createCompilationResult:
           call: http.post
           args:
               url: ${"https://dataform.googleapis.com/v1/" + repository + "/compilationResults"}
               auth:
                   type: OAuth2
               body:
                   gitCommitish: GIT_COMMITISH
           result: compilationResult
       - createWorkflowInvocation:
           call: http.post
           args:
               url: ${"https://dataform.googleapis.com/v1/" + repository + "/workflowInvocations"}
               auth:
                   type: OAuth2
               body:
                   compilationResult: ${compilationResult.body.name}
           result: workflowInvocation
       - complete:
           return: ${workflowInvocation.body.name}
   

   מחליפים את מה שכתוב בשדות הבאים:

   • ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud .
   • ‫REPOSITORY_LOCATION: המיקום של מאגר Dataform.
   • ‫REPOSITORY_ID: השם של מאגר Dataform.
   • ‫GIT_COMMITISH: ה-commitish של Git, כמו branch,‏ tag או commit SHA, שממנו רוצים להריץ את קוד Dataform. במאגר חדש שנוצר, מחליפים את הערך הזה ב-main.

3. תזמון תהליך העבודה של התזמור באמצעות Cloud Scheduler.

התאמה אישית של הבקשה ליצירת תוצאת הידור בתהליך העבודה של Dataform

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

לדוגמה, כדי להוסיף הגדרה של _dev schemaSuffix לכל הפעולות במהלך ההידור, מחליפים את גוף השלב createCompilationResult בקטע הקוד הבא:

    - createCompilationResult:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1/" + repository + "/compilationResults"}
            auth:
                type: OAuth2
            body:
                gitCommitish: GIT_COMMITISH
                codeCompilationConfig:
                    schemaSuffix: dev

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

התאמה אישית של הבקשה להפעלת תהליך העבודה ב-Dataform

אתם יכולים לעדכן את תהליך העבודה הקיים של התזמור ולהגדיר את הגדרות הבקשה להפעלת תהליך העבודה של Dataform בפורמט YAML. למידע נוסף על ההגדרות של בקשת ההפעלה, קראו את ההפניה למשאב REST של projects.locations.repositories.workflowInvocations.

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

    - createWorkflowInvocation:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1/" + repository + "/workflowInvocations"}
            auth:
                type: OAuth2
            body:
                compilationResult: ${compilationResult.body.name}
                invocationConfig:
                    includedTags:
                    - hourly
                    transitiveDependenciesIncluded: true
                

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

תזמון הרצות באמצעות Managed Airflow

אתם יכולים להשתמש ב-Managed Airflow 2 כדי לתזמן הפעלות של Dataform. ב-Dataform אין תמיכה ב-Managed Airflow 1.

כדי לנהל את לוחות הזמנים של הפעלות Dataform באמצעות Managed Airflow 2, אתם יכולים להשתמש באופרטורים של Dataform בגרפים אציקליים מכוונים (DAG) של Airflow. אתם יכולים ליצור DAG של Airflow שמתזמן הפעלות של תהליכי עבודה ב-Dataform.

ב-Dataform יש אופרטורים שונים של Airflow. הם כוללים אופרטורים לקבלת תוצאת הידור, לקבלת הפעלה של תהליך עבודה ולביטול הפעלה של תהליך עבודה. הרשימה המלאה של האופרטורים של Dataform Airflow זמינה במאמר Google Dataform Operators.

הערה: כשמתזמנים הפעלות של תהליכי עבודה באמצעות Managed Airflow, כדי להפעיל את DataformCreateCompilationResultOperator עם ערך git_commitish, כמו הסתעפות, תג או SHA של commit, צריך לשכפל את מאגר Git המרוחק דרך הרשת בכל הפעלה של Dataform. חיבורים לרשתות חיצוניות יכולים להיות לא יציבים באופן זמני במהלך קומפילציות תכופות או בו-זמניות. לכן, מומלץ לתזמן תהליכי עבודה בסביבת הייצור באמצעות הגדרות גרסה, או להגדיר ניסיונות חוזרים אוטומטיים עם השהיה מעריכית לפני ניסיון חוזר (exponential backoff) באופרטור. מידע נוסף זמין במאמר לא ניתן לגשת למאגר המרוחק.

מתקינים את חבילת PyPI של google-cloud-dataform

אם אתם משתמשים בגרסאות 2.0.25 ואילך של Managed Airflow 2, החבילה הזו מותקנת מראש בסביבה שלכם. אין צורך להתקין אותה.

חשוב: בתמונות של Managed Airflow עם Airflow 2.3.3 נעשה שימוש בגרסה הציבורית 8.1.0 של החבילה apache-airflow-providers-google. גרסה 8.1.0 לא כוללת מחברים של Dataform. התמיכה במחברים של Dataform תהיה זמינה החל מגרסה 8.3.0, שתשוחרר יחד עם קובצי אימג' של Managed Airflow עם Airflow 2.3.4. מידע נוסף זמין בנתוני הגרסה של Managed Airflow.

אם אתם משתמשים בגרסאות קודמות של Managed Airflow 2, התקינו את חבילת PyPI של google-cloud-dataform.

בקטע PyPI packages, מציינים את הגרסה ==0.2.0.

יצירת DAG של Airflow שמתזמן הפעלות של תהליכי עבודה ב-Dataform

כדי לנהל הפעלות מתוזמנות של תהליכי עבודה ב-Dataform באמצעות Managed Airflow 2, כותבים את ה-DAG באמצעות אופרטורים של Dataform Airflow, ואז מעלים אותו לקטגוריה של הסביבה.

דוגמת הקוד הבאה מציגה DAG של Airflow שיוצר תוצאת הידור של Dataform ומתחיל הפעלה של תהליך עבודה ב-Dataform:

from datetime import datetime

from airflow import models
from airflow.models.baseoperator import chain
from airflow.providers.google.cloud.operators.dataform import (
    DataformCreateCompilationResultOperator,
    DataformCreateWorkflowInvocationOperator,
)

DAG_ID = "dataform"
PROJECT_ID = "PROJECT_ID"
REPOSITORY_ID = "REPOSITORY_ID"
REGION = "REGION"
GIT_COMMITISH = "GIT_COMMITISH"

with models.DAG(
    DAG_ID,
    schedule_interval='@once',  # Override to match your needs
    start_date=datetime(2022, 1, 1),
    catchup=False,  # Override to match your needs
    tags=['dataform'],
) as dag:

    create_compilation_result = DataformCreateCompilationResultOperator(
        task_id="create_compilation_result",
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        compilation_result={
            "git_commitish": GIT_COMMITISH,
        },
    )
    create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
        task_id='create_workflow_invocation',
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
         workflow_invocation={
            "compilation_result": "{{ task_instance.xcom_pull('create_compilation_result')['name'] }}"
        },
    )


create_compilation_result >> create_workflow_invocation

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט ב-Dataform Google Cloud .
• ‫REPOSITORY_ID: השם של מאגר Dataform.
• ‫REGION: האזור שבו נמצא מאגר Dataform.
• ‫COMPILATION_RESULT: השם של תוצאת ה-compilation שבה רוצים להשתמש להפעלה הזו של תהליך העבודה.
• ‫GIT_COMMITISH: ה-commitish של Git, כמו הסתעפות, תג או SHA של commit, של גרסת הקוד במאגר ה-Git המרוחק שרוצים לקמפל.

בדוגמת הקוד הבאה מוצג DAG של Airflow שמבצע את הפעולות הבאות:

1. יצירת תוצאת הידור (compilation) ב-Dataform.
2. הפעלת תהליך עבודה אסינכרוני ב-Dataform.
3. דגימת הסטטוס של תהליך העבודה עד שהוא נכנס למצב הצפוי באמצעות DataformWorkflowInvocationStateSensor.

from datetime import datetime

from google.cloud.dataform_v1 import WorkflowInvocation

from airflow import models
from airflow.models.baseoperator import chain
from airflow.providers.google.cloud.operators.dataform import (
    DataformCreateCompilationResultOperator,
    DataformCreateWorkflowInvocationOperator,
)
from airflow.providers.google.cloud.sensors.dataform import DataformWorkflowInvocationStateSensor

DAG_ID = "dataform"
PROJECT_ID = "PROJECT_ID"
REPOSITORY_ID = "REPOSITORY_ID"
REGION = "REGION"
GIT_COMMITISH = "GIT_COMMITISH"

with models.DAG(
    DAG_ID,
    schedule_interval='@once',  # Override to match your needs
    start_date=datetime(2022, 1, 1),
    catchup=False,  # Override to match your needs
    tags=['dataform'],
) as dag:

    create_compilation_result = DataformCreateCompilationResultOperator(
        task_id="create_compilation_result",
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        compilation_result={
            "git_commitish": GIT_COMMITISH,
        },
    )

create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
    task_id='create_workflow_invocation',
    project_id=PROJECT_ID,
    region=REGION,
    repository_id=REPOSITORY_ID,
    asynchronous=True,
    workflow_invocation={
        "compilation_result": COMPILATION_RESULT
    }
)

is_workflow_invocation_done = DataformWorkflowInvocationStateSensor(
    task_id="is_workflow_invocation_done",
    project_id=PROJECT_ID,
    region=REGION,
    repository_id=REPOSITORY_ID,
    workflow_invocation_id=("{{ task_instance.xcom_pull('create_workflow_invocation')['name'].split('/')[-1] }}"),
    expected_statuses={WorkflowInvocation.State.SUCCEEDED},
)


create_compilation_result >> create_workflow_invocation

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט ב-Dataform Google Cloud .
• ‫REPOSITORY_ID: השם של מאגר Dataform.
• ‫REGION: האזור שבו נמצא מאגר Dataform.
• ‫COMPILATION_RESULT: השם של תוצאת ה-compilation שבה רוצים להשתמש להפעלה הזו של תהליך העבודה.
• ‫GIT_COMMITISH: ה-commitish של Git, כמו הסתעפות, תג או SHA של commit, של גרסת הקוד במאגר ה-Git המרוחק שרוצים לקמפל.

הוספת פרמטרים של הגדרות אישיות של הידור

אפשר להוסיף עוד פרמטרים של הגדרות קומפילציה לאובייקט DAG של create_compilation_result ב-Airflow. מידע נוסף על הפרמטרים הזמינים מופיע במאמרי העזרה של CodeCompilationConfig Dataform API.

• כדי להוסיף פרמטרים של הגדרות קומפילציה לאובייקט Airflow DAG של create_compilation_result, מוסיפים את הפרמטרים שנבחרו לשדה code_compilation_config בפורמט הבא:

      create_compilation_result = DataformCreateCompilationResultOperator(
          task_id="create_compilation_result",
          project_id=PROJECT_ID,
          region=REGION,
          repository_id=REPOSITORY_ID,
          compilation_result={
              "git_commitish": GIT_COMMITISH,
              "code_compilation_config": { "PARAMETER": "PARAMETER_VALUE"}
          },
      )
  

  מחליפים את מה שכתוב בשדות הבאים:

  • ‫PROJECT_ID: מזהה הפרויקט ב-Dataform Google Cloud .
  • ‫REPOSITORY_ID: השם של מאגר Dataform.
  • ‫REGION: האזור שבו נמצא המאגר של Dataform.
  • ‫GIT_COMMITISH: ה-commitish של Git, כמו הסתעפות, תג או SHA של commit, של גרסת הקוד במאגר ה-Git המרוחק שרוצים לקמפל.
  • ‫PARAMETER: הפרמטר CodeCompilationConfig שנבחר. אפשר להוסיף כמה פרמטרים.
  • PARAMETER_VALUE: הערך של הפרמטר שנבחר.

בדוגמת הקוד הבאה מוצג הפרמטר defaultDatabase שנוסף לאובייקט create_compilation_result Airflow DAG:

    create_compilation_result = DataformCreateCompilationResultOperator(
        task_id="create_compilation_result",
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        compilation_result={
            "git_commitish": REMOTE_BRANCH,
            "code_compilation_config": { "default_database": "my-custom-gcp-project"}
        },
    )

הוספת פרמטרים להגדרת הפעלה של תהליך עבודה

אפשר להוסיף עוד פרמטרים של הגדרת הפעלה של תהליך עבודה לאובייקט DAG של create_workflow_invocation ב-Airflow. מידע נוסף על הפרמטרים הזמינים מופיע במאמרי העזרה של InvocationConfig Dataform API.

• כדי להוסיף פרמטרים של הגדרות להפעלה של תהליך עבודה לאובייקט create_workflow_invocation Airflow DAG, מוסיפים את הפרמטרים שנבחרו לשדה invocation_config בפורמט הבא:

      create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
          task_id='create_workflow_invocation',
          project_id=PROJECT_ID,
          region=REGION,
          repository_id=REPOSITORY_ID,
          workflow_invocation={
              "compilation_result": "{{ task_instance.xcom_pull('create_compilation_result')['name'] }}",
              "invocation_config": { "PARAMETER": PARAMETER_VALUE }
          },
      )
  
  

  מחליפים את מה שכתוב בשדות הבאים:

  • ‫PROJECT_ID: מזהה הפרויקט ב-Dataform Google Cloud .
  • ‫REPOSITORY_ID: השם של מאגר Dataform.
  • ‫REGION: האזור שבו נמצא המאגר של Dataform.
  • ‫PARAMETER: הפרמטר InvocationConfig שנבחר. אפשר להוסיף כמה פרמטרים.
  • PARAMETER_VALUE: הערך של הפרמטר שנבחר.

בדוגמת הקוד הבאה מוצגים הפרמטרים includedTags[] ו-transitiveDependenciesIncluded שנוספו לאובייקט create_workflow_invocation Airflow DAG:

    create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
        task_id='create_workflow_invocation',
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        workflow_invocation={
            "compilation_result": "{{ task_instance.xcom_pull('create_compilation_result')['name'] }}",
            "invocation_config": { "included_tags": ["daily"], "transitive_dependencies_included": true }
        },
    )

אוטומציה של הפעלות באמצעות טריגרים לפיתוח גרסת Build ב-Cloud

אם אתם רוצים לעבור מעבר ללוחות הזמנים שמבוססים על זמן בהגדרת הגרסה, אתם יכולים להשתמש בטריגר לפיתוח גרסת Build של Cloud Build כדי ליצור פייפליין מבוסס-אירועים. בגישה הזו, הקוד עובר קומפילציה אוטומטית בכל פעם ש-commit חדש נשלח ל-Git branch, ומיד מופעלת קריאה לתהליך עבודה ב-Dataform כדי לעדכן את הנתונים.

הכנת המשאבים

1. בפרויקט ב- Google Cloud , מפעילים את ממשקי ה-API של Dataform ו-Cloud Build:

   הפעלת Dataform API

   הפעלת Cloud Build API

2. ודאו שיש לכם:

   • חשבון שירות בהתאמה אישית שמשמש לקומפילציה. שימו לב לכתובת האימייל של חשבון השירות, לדוגמה dataform-compiler@PROJECT_NUMBER.iam.gserviceaccount.com.

   • מאגר Dataform שמחובר לספק Git.

   • הגדרת גרסה במאגר של Dataform. חשוב לזכור את מזהה ההגדרה של הגרסה.

   • הגדרת תהליך עבודה במאגר Dataform שמשתמשת בהגדרת הגרסה. שימו לב למזהה ההגדרה של תהליך העבודה.

הענקת הרשאות ה-IAM הנדרשות

מקצים את התפקיד "אדמין ב-Dataform" (roles/dataform.admin) לחשבון השירות המותאם אישית במאגר של Dataform. התפקיד הזה נותן גישה מלאה למאגר, כולל הרשאה ליצור תוצאות קומפילציה, לעדכן הגדרות של גרסאות ולהפעיל קריאות חדשות לתהליכי עבודה. במאמר שליטה בגישה למאגר ספציפי מוסבר איך מקצים תפקיד IAM למאגר ספציפי.

מקצים את התפקיד'משתמש בחשבון שירות' (roles/iam.serviceAccountUser) בחשבון השירות בהתאמה אישית של הגדרת תהליך העבודה, לחשבון השירות של הטריגר לפיתוח גרסת Build של Cloud Build. מידע נוסף על הדרישה הזו זמין במאמר שימוש במצב act-as מחמיר.

כדי ש-Cloud Build יוכל להשתמש בחשבון השירות בהתאמה אישית, צריך לתת לסוכן השירות של Cloud Build הרשאה לפעול בשם החשבון הזה. כדי לתת לסוכן השירות של Cloud Build הרשאת התחזות:

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

   לדף Service accounts

2. בוחרים את חשבון השירות בהתאמה אישית.

3. עוברים לכרטיסייה Principals with access.

4. לוחצים על Grant Access.

5. בשדה New principals, מזינים את כתובת האימייל של סוכן השירות של Cloud Build, בפורמט הבא:

   [e-mail]
   

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

6. בתפריט Select a role, בוחרים באפשרות Service Account User.

7. לוחצים על Save.

יצירת קובץ התצורה של cloudbuild.yaml

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

steps:
  - name: 'gcr.io/google.com/cloudsdktool/cloud-sdk:latest'
    entrypoint: 'bash'
    args:
      - '-c'
      - |
        set -e -o pipefail # Exit script on any error

        # 1. Get the access token
        TOKEN=$(gcloud auth print-access-token)

        # 2. Define API endpoints and resource names
        RELEASE_CONFIG_RESOURCE="projects/${_PROJECT_ID}/locations/${_DATAFORM_LOCATION}/repositories/${_DATAFORM_REPO_ID}/releaseConfigs/${_RELEASE_CONFIG_ID}"
        COMPILATION_RESULTS_API="https://dataform.googleapis.com/v1/projects/${_PROJECT_ID}/locations/${_DATAFORM_LOCATION}/repositories/${_DATAFORM_REPO_ID}/compilationResults"

        # 3. Create the new compilation result
        echo "Creating new compilation result from $$RELEASE_CONFIG_RESOURCE..."
        CREATE_PAYLOAD="{\"releaseConfig\": \"$$RELEASE_CONFIG_RESOURCE\"}"
        curl --fail-with-body -X POST \
          -H "Authorization: Bearer $$TOKEN" \
          -H "Content-Type: application/json" \
          -d "$$CREATE_PAYLOAD" \
          "$$COMPILATION_RESULTS_API" | tee /workspace/compilation_response.json

  - name: 'alpine'
    entrypoint: 'bash'
    args:
      - '-c'
      - |
        set -e # Exit script on any error

        # 4. Parse compilation result name
        apk add --no-cache jq
        COMPILATION_NAME=$(jq -r '.name' < /workspace/compilation_response.json)

        echo "Successfully created compilation result: $$COMPILATION_NAME"
        echo $$COMPILATION_NAME > /workspace/compilation_result_name.txt

  - name: 'gcr.io/google.com/cloudsdktool/cloud-sdk:latest'
    entrypoint: 'bash'
    args:
      - '-c'
      - |
        set -e # Exit script on any error
        # 5. Update the releaseConfig to set the new compilation result as 'live'
        COMPILATION_NAME=$(cat /workspace/compilation_result_name.txt)
        echo "Updating release config to set $$COMPILATION_NAME as live..."
        PATCH_PAYLOAD="{\"releaseCompilationResult\": \"$$COMPILATION_NAME\", \"gitCommitish\": \"$BRANCH_NAME\"}"

        RELEASE_CONFIG_RESOURCE="projects/${_PROJECT_ID}/locations/${_DATAFORM_LOCATION}/repositories/${_DATAFORM_REPO_ID}/releaseConfigs/${_RELEASE_CONFIG_ID}"
        RELEASE_CONFIG_PATCH_API="https://dataform.googleapis.com/v1/$${RELEASE_CONFIG_RESOURCE}"
        curl --fail-with-body -X PATCH \
          -H "Authorization: Bearer $(gcloud auth print-access-token)" \
          -H "Content-Type: application/json" \
          -d "$$PATCH_PAYLOAD" \
          "$$RELEASE_CONFIG_PATCH_API?updateMask=releaseCompilationResult"

        echo "Successfully updated release config."

  - name: 'gcr.io/google.com/cloudsdktool/cloud-sdk:latest'
    entrypoint: 'bash'
    args:
      - '-c'
      - |
        set -e # Exit script on any error
        # 6. Launch a workflow config after recompiling the release config
        WORKFLOW_CONFIG_RESOURCE="projects/${_PROJECT_ID}/locations/${_DATAFORM_LOCATION}/repositories/${_DATAFORM_REPO_ID}/workflowConfigs/${_WORKFLOW_CONFIG_ID}"
        CREATE_WORKFLOW_PAYLOAD="{\"workflowConfig\": \"$$WORKFLOW_CONFIG_RESOURCE\"}"

        WORKFLOW_INVOCATIONS_API="https://dataform.googleapis.com/v1/projects/${_PROJECT_ID}/locations/${_DATAFORM_LOCATION}/repositories/${_DATAFORM_REPO_ID}/workflowInvocations"
        curl --fail-with-body -X POST \
          -H "Authorization: Bearer $(gcloud auth print-access-token)" \
          -H "Content-Type: application/json" \
          -d "$$CREATE_WORKFLOW_PAYLOAD" \
          "$$WORKFLOW_INVOCATIONS_API"

        echo "Successfully created a new workflow invocation."

# Define substitution variables that can be set in the trigger
substitutions:
  _DATAFORM_LOCATION: 'us-central1'  # Default, change if needed
  _DATAFORM_REPO_ID: ''              # Required: Set this in the trigger
  _RELEASE_CONFIG_ID: ''             # Required: Set this in the trigger
  _WORKFLOW_CONFIG_ID: ''            # Required: Set this in the trigger
  _PROJECT_ID: ${PROJECT_ID}         # Automatically uses the build's Project ID

options:
  logging: CLOUD_LOGGING_ONLY

יצירת הטריגר לפיתוח גרסת Build ב-Cloud

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

1. במסוף Google Cloud , פותחים את הדף Triggers של Cloud Build.

   כניסה לדף Triggers

2. אם עדיין לא קישרתם את מאגר ה-Git, לוחצים על Connect repository ופועלים לפי השלבים.

3. לוחצים על Create trigger (יצירת טריגר).

4. כותבים שם לטריגר.

5. בוחרים אזור לטריגר.

6. בוחרים אירוע לטריגר.

7. בקטע Source, מגדירים את המאגר למאגר Git המקושר.

8. מגדירים את ההסתעפות כהסתעפות הראשית של המאגר.

9. בקטע Configuration, בוחרים את קובץ התצורה של Cloud Build, שיכול להיות קובץ YAML או JSON.

10. מגדירים את מיקום הקובץ בתור /cloudbuild.yaml או הנתיב לקובץ.

11. בקטע Substitution variables, מוסיפים את המשתנים והערכים הבאים:

    • ‫_DATAFORM_REPO_ID: מזהה המאגר ב-Dataform
    • ‫_RELEASE_CONFIG_ID: מזהה הגדרת הגרסה ב-Dataform
    • ‫_WORKFLOW_CONFIG_ID: מזהה הגדרת תהליך העבודה ב-Dataform
    • אופציונלי: _DATAFORM_LOCATION: האזור של מאגר Dataform, לדוגמה, us-central1

12. בקטע Service Account, בוחרים את חשבון השירות בהתאמה אישית.

13. לוחצים על יצירה.

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

בדיקת הטריגר

1. מבצעים commit ו-push לקובץ cloudbuild.yaml לענף שהטריגר עוקב אחריו.

2. כדי לראות את ה-build של Cloud Build, פותחים את הדף Build History במסוף Google Cloud .

   כניסה לדף Build History

3. אם ה-build בוצע בהצלחה, נכנסים לדף Dataform.

   כניסה ל-Dataform

4. בוחרים את המאגר.

5. לוחצים על גרסאות ותזמון ובוחרים את הגדרת הגרסה.

6. ברשימה Manual / API compilation results, מחפשים רשומה חדשה. הקומפילציה האחרונה שהושלמה בהצלחה צריכה להיות מסומנת בתור Live compilation result בהגדרת הגרסה.

7. לוחצים על Workflow execution logs.

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

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

• במאמר יצירת הגדרת גרסה מוסבר איך מגדירים את גרסאות האוספים ב-Dataform.
• מבוא למחזור החיים של הקוד ב-Dataform
• מידע נוסף על Dataform API
• סקירה כללית על סביבות Managed Airflow
• למחירון של Workflows