תזמון הרצות

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

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

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

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

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

כדי לתזמן הפעלות עם הגדרות של תהליכי עבודה או לתזמן הפעלות עם תהליכי עבודה ו-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 כשיוצרים הגדרת תהליך עבודה (גרסת Preview), צריך להעניק גישה לחשבון Google.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

• פרטי הכניסה של משתמש בחשבון Google (גרסת טרום-השקה (Preview))

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

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

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

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

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

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

1. במאגר, עוברים אל Releases & Scheduling (פרסומים ותזמון).
2. בקטע הגדרות של תהליכי עבודה, לוחצים על יצירה.

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

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

4. בתפריט Release configuration (הגדרת גרסה), בוחרים הגדרת גרסת קומפילציה.

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

   • כדי להשתמש בפרטי הכניסה של המשתמש בחשבון Google (תצוגה מקדימה), בוחרים באפשרות Execute with my user credentials (הפעלה עם פרטי הכניסה של המשתמש).
   • כדי להשתמש בחשבון שירות בהתאמה אישית, בוחרים באפשרות Execute with selected service account ואז בוחרים את חשבון השירות שמשויך לפרויקט Google Cloud שיש לכם גישה אליו. אם לא בוחרים חשבון שירות, הגדרת תהליך העבודה משתמשת בחשבון השירות של המאגר.
   הערה: אי אפשר להריץ תהליכי עבודה עם סוכן השירות שמוגדר כברירת מחדל ב-Dataform. אתם צריכים להשתמש בחשבון שירות בהתאמה אישית או בפרטי הכניסה של המשתמש בחשבון Google.

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

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

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

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

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

   • כדי להפעיל את כל תהליך העבודה, לוחצים על כל הפעולות.
   • כדי להפעיל פעולות נבחרות בתהליך העבודה, לוחצים על בחירת פעולות ואז בוחרים פעולות.
   • כדי לבצע פעולות בתגים שנבחרו, לוחצים על בחירת תגים ואז בוחרים תגים.
   • אופציונלי: כדי להפעיל תגים או פעולות נבחרות ואת התלויות שלהם, בוחרים באפשרות Include dependencies (כולל תלויות).
   • אופציונלי: כדי להפעיל את הפעולות או התגים שנבחרו ואת התגים שתלויים בהם, בוחרים באפשרות Include dependents (כולל תגים שתלויים בתגים שנבחרו).

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

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

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

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

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

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

איך נותנים הרשאה לחשבון Google

תצוגה מקדימה

השימוש במוצר או בתכונה כפוף לתנאי השימוש במוצרים בגרסת טרום-GA, בחלק "תנאים כלליים לשירות" של התנאים הספציפיים לשירות. השימוש בתכונות ובמוצרים בגרסת טרום-GA הוא "כפי שהם" (As is), ויכול להיות שהתמיכה בהם תהיה מוגבלת. מידע נוסף זמין בקטע תיאור שלבי ההשקה.

הערה: כדי לבקש תמיכה או לשלוח משוב על התכונה הזו, אפשר לפנות אל [e-mail].

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

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

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

כדי לבטל את ההרשאה שנתתם:

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

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

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

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

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

1. במאגר, עוברים אל Releases & Scheduling (פרסומים ותזמון).
2. לצד הגדרת תהליך העבודה שרוצים לערוך, לוחצים על התפריט עוד ואז על עריכה.
3. בחלונית עריכת הגדרות תהליך העבודה, עורכים את ההגדרות של תהליך העבודה ולוחצים על שמירה.

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

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

1. במאגר, עוברים אל Releases & Scheduling (פרסומים ותזמון).
2. לצד הגדרת תהליך העבודה שרוצים למחוק, לוחצים על התפריט אפשרויות נוספות ואז על מחיקה.
3. בתיבת הדו-שיח מחיקת הגדרת הפצה, לוחצים על מחיקה.

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

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

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

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

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

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

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

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

   • תפקיד עורך ב-Dataform (roles/dataform.editor).
   • התפקיד Service Account User (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: הענף ב-Git שממנו רוצים להריץ את קוד Dataform. אם המאגר נוצר לאחרונה, מחליפים את main.

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

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

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

לדוגמה, כדי להוסיף הגדרה של _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. מידע נוסף על ההגדרות של בקשת ההפעלה זמין בprojects.locations.repositories.workflowInvocations הפניית משאבי REST.

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

    - 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.

התקנת חבילת google-cloud-dataform PyPI

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

חשוב: תמונות 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, צריך להתקין את חבילת google-cloud-dataform PyPI.

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

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

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

בדוגמת הקוד הבאה מוצג Airflow DAG שיוצר תוצאת קומפילציה של 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: השם של תוצאת הקומפילציה שרוצים להשתמש בה להפעלת תהליך העבודה הזה.
• ‫GIT_COMMITISH: ה-Git commitish במאגר Git מרוחק של גרסת הקוד שרוצים להשתמש בה – לדוגמה, ענף או Git SHA.

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

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: השם של תוצאת הקומפילציה שרוצים להשתמש בה להפעלת תהליך העבודה הזה.
• ‫GIT_COMMITISH: ה-Git commitish במאגר Git מרוחק של גרסת הקוד שרוצים להשתמש בה – לדוגמה, ענף או Git SHA.
• ‫COMPILATION_RESULT: השם של תוצאת הקומפילציה שרוצים להשתמש בה להפעלת תהליך העבודה הזה.

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

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

• כדי להוסיף פרמטרים של הגדרת קומפילציה לאובייקט create_compilation_resultAirflow DAG, מוסיפים את הפרמטרים שנבחרו לשדה 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: ה-Git commitish במאגר Git מרוחק של גרסת הקוד שרוצים להשתמש בה – לדוגמה, ענף או Git SHA.
  • ‫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"}
        },
    )

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

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

• כדי להוסיף פרמטרים להגדרת הפעלת זרימת העבודה לאובייקט 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 }
        },
    )

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

אם אתם רוצים לעבור מעבר ללוחות זמנים מבוססי-זמן בהגדרת הפצה, אתם יכולים להשתמש בטריגר לפיתוח גרסת Build של Cloud Build כדי ליצור פייפליין מבוסס-אירועים. בגישה הזו, הקוד עובר קומפילציה באופן אוטומטי בכל פעם שמתבצעת העברה (push) של קומיט חדש לענף Git, ומופעלת מיד קריאה (invocation) של תהליך עבודה ב-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 את התפקיד Dataform Admin ‏(roles/dataform.admin). התפקיד הזה מספק גישה מלאה למאגר, כולל ההרשאה ליצור תוצאות קומפילציה, לעדכן הגדרות של גרסאות ולבצע הפעלות חדשות של תהליכי עבודה. פרטים על מתן תפקיד IAM למאגר ספציפי זמינים במאמר שליטה בגישה למאגר ספציפי.

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

כדי ש-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

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

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

   כניסה לדף Triggers

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

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

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

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

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

7. בקטע מקור, מגדירים את המאגר למאגר 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. אם הבנייה בוצעה בהצלחה, עוברים לדף Dataform.

   מעבר אל Dataform

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

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

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

7. לוחצים על Workflow execution logs (יומני ביצוע של תהליכי עבודה).

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

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

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