פיתוח צינורות עיבוד נתונים

במדריך הזה מוסבר איך ליצור ולפרוס צינור עיבוד נתונים לארגון נתונים באמצעות Google Cloud Data Agent Kit.

צינור הנתונים לדוגמה מריץ סקריפט PySpark ב-Managed Service for Apache Spark.

אפשר לפרוס Orchestration Pipelines מתוך סביבת הפיתוח המשולבת (IDE) כגרסאות מקומיות או באמצעות פעולת GitHub, למשל כשממזגים שינויים עם הענף main. במאמר הזה נסביר איך פורסים את הגרסה המקומית של Orchestration Pipelines.

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

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

  1. התקנת Data Agent Kit
  2. קובעים את ההגדרות.
  3. מוסיפים מאגר מ-GitHub לסביבת העבודה של ה-IDE כדי לאחסן Orchestration Pipelines ונכסים כמו סקריפטים.

בדיקת התפקידים הנדרשים ב-IAM

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

כדי ליצור ולנהל סביבות של Managed Service for Apache Airflow ולנהל אובייקטים בדליים המשויכים, צריך את התפקידים הבאים. מידע נוסף על תפקידי המשתמשים האלה זמין במאמר הענקת תפקידים למשתמשים במסמכי התיעוד של Managed Service for Apache Airflow.

  • אדמין של סביבה ואובייקט אחסון (composer.environmentAndStorageObjectAdmin)
  • משתמש בחשבון שירות (iam.serviceAccountUser)

כדי לעבוד עם משאבים של BigQuery ו-Cloud Storage, אתם צריכים את התפקידים הבאים.

  • עריכה של נתוני BigQuery (roles/bigquery.dataEditor)
  • אדמין של אובייקטים באחסון (roles/storage.objectAdmin)

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

יצירת חשבון שירות והקצאת תפקידי IAM לחשבון

משתמשים בחשבון שירות ייחודי לסביבת Managed Airflow Gen 3. חשבון השירות יוצר סביבת Managed Airflow מדור 3 ומריץ את כל Orchestration Pipelines שאתם פורסים.

צריך לבקש מהאדמין לבצע את השלבים הבאים:

  1. יוצרים חשבון שירות כמו שמתואר במסמכי ה-IAM.
  2. מקצים לחשבון השירות את התפקיד Composer Worker (composer.worker). ברוב המקרים, התפקיד הזה מספק את ההרשאות הנדרשות.

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

יצירת Google Cloud משאבים לצינור עיבוד הנתונים של התזמור

בשלב הזה יוצרים Google Cloud משאבים לצינור האורקסטרציה.

יצירת סביבת Managed Airflow מדור 3

יוצרים סביבת Managed Airflow מדור 3 עם ההגדרות הבאות:

  • שם הסביבה: מזינים שם שישמש בהמשך להגדרת צינור הנתונים של תזמור התהליכים. לדוגמה, example-pipeline-scheduler.
  • מיקום: בוחרים מיקום. מומלץ ליצור את כל המשאבים במדריך הזה באותו מיקום. לדוגמה, us-central1.
  • חשבון שירות: בוחרים את חשבון השירות שיצרתם עבור הסביבה הזו.

בדוגמה הבאה של פקודת Google Cloud CLI אפשר לראות את התחביר:

gcloud composer environments create example-pipeline-scheduler \
  --location us-central1 \
  --image-version composer-3-airflow-2 \
  --service-account "example-account@example-project.iam.gserviceaccount.com"

הוספת פרמטרים של סביבה להגדרות של כלי התזמון

צריך לספק פרטי חיבור לסביבת Managed Airflow שבה יופעל צינור הנתונים של תזמור התהליכים.

מוסיפים את פרמטרי ההגדרה של הסביבה שיצרתם באמצעות הכלי לעריכת הגדרות של Google Cloud Data Agent Kit:

  1. לוחצים על הסמל Google Cloud Data Agent Kit בסרגל הפעילות.
  2. מרחיבים את הגדרות ולוחצים על הגדרות.
  3. לוחצים על כלי התזמון.
  4. מזינים את הפרמטרים של סביבת Managed Airflow Gen 3 שיצרתם קודם:
    • מזהה הפרויקט: שם הפרויקט שבו נמצאת הסביבה. לדוגמה: example-project.
    • אזור: האזור שבו נמצאת הסביבה. דוגמה: us-central1.
    • סביבה: שם הסביבה. דוגמה: example-pipeline-scheduler.
  5. לוחצים על Save.

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

יוצרים קטגוריה של Cloud Storage באותו פרויקט שבו נמצא סביבת Managed Airflow, ונותנים לה שם דומה ל-example-pipelines-bucket. הבאקט הזה נדרש כדי לאחסן את המשימה שלכם ב-Managed Service for Apache Spark.

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

יצירת מערך נתונים וטבלה חדשים ב-BigQuery

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

  1. יוצרים קבוצת נתונים חדשה בשם wordcount_dataset.
  2. יוצרים טבלה חדשה ב-BigQuery בשם wordcount_output.

הוספת נכסי צינור

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

לא אג'נטי

מוסיפים את הקובץ הבא לתיקייה /scripts במאגר. בשלב מאוחר יותר מוסיפים פעולת צינור שמעבדת את הסקריפט הזה ב-Managed Service for Apache Spark.

קובץ wordcount.py לדוגמה:

#!/usr/bin/python
"""BigQuery I/O PySpark example for Word Count"""

from pyspark.sql import SparkSession

spark = SparkSession \
.builder \
.appName('spark-bigquery-demo') \
.getOrCreate()

# Use the Cloud Storage bucket for temporary BigQuery export data used
# by the connector.
bucket = ARTIFACTS_BUCKET_NAME
spark.conf.set('temporaryGcsBucket', bucket)

# Load data from BigQuery public dataset (Shakespeare).
words = spark.read.format('bigquery') \
.option('table', 'bigquery-public-data:samples.shakespeare') \
.load()
words.createOrReplaceTempView('words')

# Perform word count using Spark SQL.
# This query counts occurrences of each word.
word_count = spark.sql(
    'SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word ORDER BY word_count DESC'
)
word_count.show()
word_count.printSchema()

# Saving the results to a new table in BigQuery.
# Replace YOUR_PROJECT_ID with your project ID.
destination_table = 'PROJECT_ID:wordcount_dataset.wordcount_output'
word_count.write.format('bigquery') \
.option('table', destination_table) \
.mode('overwrite') \
.save()

print(f"Successfully wrote word counts to BigQuery table: {destination_table}")

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

  • ‫ARTIFACTS_BUCKET_NAME: השם של הקטגוריה ב-Cloud Storage שיצרתם קודם. לדוגמה: example-pipelines-bucket.
  • ‫PROJECT_ID: שם הפרויקט שבו נמצאת הסביבה. לדוגמה: example-project.

פעולות סוכן

מבקשים מהסוכן ליצור סקריפט PySpark לדוגמה בתיקייה /scripts במאגר. בהמשך תוסיפו פעולת צינור שפועלת בסקריפט הזה ב-Managed Service for Apache Spark.

כותבים פרומפט שדומה לזה:

I want to create a PySpark script that does the following:

1. Loads data from the bigquery-public-data:samples.shakespeare.
2. Counts occurrences of each word across all works using a Spark SQL query.
Sum the existing word counts for each word to get the total occurrences.
I want the results to be ordered by the word popularity, most popular first.
3. Saves results to a new table in BigQuery, in my project.

My project is sample-project, the destination table is
wordcount_dataset.wordcount_output, and I want to store temporary BigQuery
export data in example-pipelines-bucket.

Save the resulting script to /scripts as wordcount.py

אתחול Orchestration Pipelines במאגר

כשמפעילים Orchestration Pipelines, Data Agent Kit יוצר מסגרת שכוללת את הרכיבים הבאים:

  • קובץ YAML של Orchestration Pipelines: דוגמה להגדרת צינור שמכילה תזמון אבל לא פעולות מוגדרות.
  • ‫deployment.yaml: הגדרת פריסה לדוגמה של צינור עיבוד נתונים שמגדירה איך צריך לפרוס את צינור עיבוד הנתונים. בקובץ הזה מוצגת ההגדרה הנדרשת לסביבת Managed Airflow, למאגר הארטיפקטים ולכל משאב אחר שמשמש את פעולות צינור הנתונים.
  • ‫.github/workflows/deploy.yaml: מגדיר פעולה ב-GitHub שמבצעת פריסה של צינור הנתונים כשממזגים שינויים עם ההסתעפות main במאגר ב-GitHub.
  • ‫.github/workflows/validate.yaml: מגדיר פעולת GitHub שמאמתת את צינור הנתונים אחרי הפריסה שלו.

בהמשך המאמר הזה, נרחיב את ההגדרות האלה באמצעות Data Agent Kit כדי ליצור ולפרוס Orchestration Pipelines באופן מקומי.

לא אג'נטי

כדי לאתחל Orchestration Pipelines, בצע את הפעולות הבאות:

  1. לוחצים על הסמל Google Cloud Data Agent Kit בסרגל הפעילות.
  2. מרחיבים את Data Engineering ולוחצים על Initialize orchestration pipeline.

  3. מזינים פרמטרים לצינור התזמור החדש:

    1. מזהה הסביבה: השם של הסביבה שרוצים לפתח איתה. לדוגמה: dev/staging.
    2. מזהה הפייפליין: מזינים את מזהה הפייפליין. לדוגמה: example-pipeline.
    3. מזהה פרויקט בענן של Google: השם של הפרויקט שבו נמצאת הסביבה. לדוגמה: example-project.
    4. אזור: האזור שבו נמצאת הסביבה שלכם. דוגמה: us-central1.

    5. ‫Scheduler Managed Service for Apache Airflow Environment: השם של הסביבה שבה רוצים לתזמר את צינורות העיבוד. בפרמטר הזה מציינים את אותה סביבה שמוגדרת במסמך.

    6. ‫Artifacts Bucket (קטגוריית ארטיפקטים): שם הקטגוריה שמשמשת לארטיפקטים של צינור, בלי הקידומת gs://. לדוגמה: example-pipelines-bucket.

  4. לוחצים על Initialize (הפעלה).

  5. מציינים את סביבת העבודה שבה רוצים לאתחל את צינור הנתונים.

פעולות סוכן

מבקשים מהסוכן ליצור scaffolding עבור Orchestration Pipelines של המאגר שלכם.

כותבים פרומפט שדומה לזה:

Initialize orchestration pipelines in my repository. Don't add any actions
or schedule yet. I want to do it later.

The pipeline is my-sample-pipeline, the project ID is my-project, and the
region is us-central1.

The environment ID is my-test-environment. Use the same environment ID for
the Scheduler Managed Service.

Store pipeline artifacts in example-pipelines-bucket.

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

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

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

לא אג'נטי

כדי לערוך צינור:

  1. לוחצים על הסמל Google Cloud Data Agent Kit בסרגל הפעילות.
  2. מרחיבים את Data Engineering (הנדסת נתונים) ואז את Orchestration Pipelines (צינורות תזמור).
  3. בוחרים באפשרות example-pipeline.yaml. ייפתח עורך צינורות הנתונים של צינור הנתונים שנבחר.
  4. אופציונלי: בוחרים את הצומת Schedule trigger. אתם יכולים לשנות את לוח הזמנים של הצינור על ידי ציון ביטוי בסגנון cron ושעות התחלה וסיום של לוח הזמנים. התזמון שמוגדר כברירת מחדל לצינור עיבוד הנתונים החדש הוא 0 2 * * *, והוא פועל כל יום בשעה 2:00.

  1. להוסיף משימה חדשה. במדריך הזה מוסיפים משימת PySpark שמריצה סקריפט PySpark שהוספתם קודם:

    1. לוחצים על הוספת משימה ראשונה כדי להוסיף צומת משימה חדש.
    2. בוחרים באפשרות Execute PySpark script (הפעלת סקריפט PySpark) ובקובץ script/wordcount.py.

    החלונית Execute PySpark script תיפתח.

    1. ב-Spark Cluster Mode, בוחרים באפשרות Serverless Spark.
    2. בקטע מיקום, מציינים את המיקום שבו נמצאת הסביבה. לדוגמה: us-central1.
    3. לוחצים על Save.

פעולות סוכן

מריצים את הפרומפט הבא:

Add the wordcount.py script to the pipeline. I want to run it in Serverless
Spark every day at 1 AM. Run it in the same region where the environment that
runs my pipeline is located. Use the minimal resource profile.

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

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

כשפורסים גרסה מקומית של Orchestration Pipelines, Data Agent Kit מעלה גרסה מקומית של חבילת הפייפליין לסביבת Managed Airflow ומריץ אותה. פריסה מקומית מיועדת לשימוש בסביבת פיתוח.

הפקודה deploy (פריסה) פורסת לוח זמנים שלא מושהה. כדי למנוע את זה, אתם יכולים להשהות את לוח הזמנים באופן ידני בחלונית 'ניהול צינורות'. אפשר גם לערוך את קובץ ה-YAML של צינור העברת הנתונים כדי להוסיף הערה לבלוק triggers: - schedule או להסיר אותו.

לא אג'נטי

כדי לפרוס גרסה מקומית של Orchestration Pipelines לדוגמה, בצע את הפעולות הבאות:

  1. לוחצים על הסמל Google Cloud Data Agent Kit בסרגל הפעילות.
  2. מרחיבים את Data Engineering (הנדסת נתונים) ואז את Orchestration Pipelines (צינורות לתזמור).
  3. בוחרים באפשרות example-pipeline.yaml. ייפתח עורך צינורות הנתונים של צינור הנתונים שנבחר.
  4. בוחרים באפשרות Run pipeline (הפעלת צינור) ואז בוחרים את סביבת הפיתוח או סביבת הביניים שיצרתם קודם.

פעולות סוכן

מריצים את הפרומפט הבא:

Deploy my pipeline

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

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

  1. לוחצים על הסמל Google Cloud Data Agent Kit בסרגל הפעילות.
  2. מרחיבים את Data Engineering (הנדסת נתונים) ובוחרים באפשרות Pipelines management (ניהול צינורות).
  3. לוחצים על השם של צינור העיבוד (example-pipeline) כדי לראות את היסטור הביצוע שלו. ברשימת ההרצות של תאריך מסוים, אפשר לראות הרצות נפרדות של צינורות, ופירוט של פעולות נפרדות בכל הרצה של צינור.
  4. לוחצים על מזהה המשימה כדי לראות את יומני הביצוע של המשימה. מכיוון שתסריט PySpark לדוגמה הופעל ב-Managed Service for Apache Spark, ביומני המשימות יהיה קישור ליומני העיבוד ברצף.

פתרון בעיות בצינורות עיבוד נתונים

אם צינור הנתונים נכשל, יופיע לחצן אבחון בחלונית ניהול צינורות.

פעולות סוכן

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

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

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

  • החלת ניתוח שורש הבעיה בסביבת העבודה הנוכחית.
  • מבקשים מהסוכן ליצור ענף חדש ולהחיל בו את השינויים.
  • פותחים כרטיס תמיכה ב-Cloud Customer Care עם פרטי הניתוח של שורש הבעיה.

לקבלת עזרה בפתרון בעיות בתוסף, אפשר לעיין במאמר בנושא פתרון בעיות.

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