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

בעזרת Agent Platform Pipelines אתם יכולים לתזמר את תהליכי העבודה של למידת המכונה (ML) בלי שרת (serverless). כדי ש-Agent Platform Pipelines יוכל לתזמר את תהליך העבודה של למידת המכונה, צריך לתאר את תהליך העבודה כצינור עיבוד נתונים. פייפליינים של ML הם תהליכי עבודה ניידים וניתנים להרחבה של ML, שמבוססים על קונטיינרים ושירותי Google Cloud .

במדריך הזה מוסבר איך להתחיל לבנות פייפליינים של ML.

באיזה SDK של צינורות עיבוד נתונים כדאי להשתמש?

ב-Agent Platform Pipelines אפשר להריץ צינורות עיבוד נתונים שנוצרו באמצעות כל אחת מערכות ה-SDK הבאות:

  • ‫Kubeflow Pipelines SDK בגרסה 2.0 ואילך

  • ‫TensorFlow Extended בגרסה 0.30.0 ואילך

אם אתם משתמשים ב-TensorFlow בתהליך עבודה של למידת מכונה שמעבד טרה-בייטים של נתונים מובְנים או נתוני טקסט, מומלץ לבנות את צינור עיבוד הנתונים באמצעות TFX.

לתרחישים אחרים, מומלץ ליצור את צינור עיבוד הנתונים באמצעות Kubeflow Pipelines SDK. כשיוצרים צינור עיבוד נתונים באמצעות Kubeflow Pipelines SDK, אפשר להטמיע את תהליך העבודה על ידי יצירת רכיבים בהתאמה אישית או שימוש חוזר ברכיבים מוכנים מראש, כמו רכיבי צינור עיבוד הנתונים שלGoogle Cloud . Google Cloud רכיבי צינור עיבוד נתונים מקלים על השימוש בשירותים של Gemini Enterprise Agent Platform, כמו AutoML, בצינור עיבוד הנתונים.

במדריך הזה מוסבר איך לבנות צינורות באמצעות Kubeflow Pipelines SDK.

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

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

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

  2. מתקינים את גרסה 2 או גרסה עדכנית יותר של Kubeflow Pipelines SDK.

    pip install --upgrade "kfp>=2,<3"

  1. כדי להשתמש בלקוח Python של Gemini Enterprise Agent Platform בצינורות עיבוד הנתונים, התקינו את ספריות הלקוח של Gemini Enterprise Agent Platform בגרסה 1.7 ואילך.

  2. כדי להשתמש בשירותים של Gemini Enterprise Agent Platform בצינורות עיבוד הנתונים, התקינו את ה-SDK שלGoogle Cloud .

איך מתחילים ליצור צינור עיבוד נתונים

כדי לתזמר את תהליך העבודה של למידת המכונה ב-Agent Platform Pipelines, קודם צריך לתאר את תהליך העבודה כצינור עיבוד נתונים. בדוגמה הבאה מוסבר איך להשתמש בGoogle Cloud רכיבי צינור עיבוד הנתונים עם Gemini Enterprise Agent Platform כדי ליצור מערך נתונים, לאמן מודל באמצעות AutoML ולפרוס את המודל המאומן לצורך חיזויים.

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

איך מגדירים אימות

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

  1. יוצרים חשבון שירות:

    1. במסוף Google Cloud , נכנסים לדף Create service account.

      כניסה לדף Create service account

    2. כותבים שם בשדה Service account name.
    3. אופציונלי: בשדה Service account description, כותבים תיאור.
    4. לוחצים על יצירה.
    5. לוחצים על השדה Select a role. בקטע All roles, בוחרים באפשרות Gemini Enterprise Agent Platform ואז באפשרות Agent Platform User.

    6. לוחצים על Done כדי ליצור את חשבון השירות.

      חשוב לא לסגור את חלון הדפדפן. תצטרכו אותו בשלב הבא.

  2. יוצרים מפתח לחשבון השירות לצורך אימות:

    1. במסוף Google Cloud , לוחצים על כתובת האימייל של חשבון השירות שנוצר.
    2. לוחצים על Keys.
    3. לוחצים על Add key ואז על Create new key.
    4. לוחצים על Create. למחשב שלכם תתבצע הורדה של קובץ JSON עם המפתח.
    5. לוחצים על Close.
  3. נותנים לחשבון השירות החדש גישה לחשבון השירות שמשמש להפעלת צינורות עיבוד הנתונים.
    1. לוחצים על כדי לחזור לרשימת חשבונות השירות.

    2. לוחצים על השם של חשבון השירות שמשמש להרצת צינורות עיבוד הנתונים. מופיע הדף Service account details.

      אם פעלתם לפי ההוראות במדריך להגדרת הפרויקט ל-Gemini Enterprise Agent Platform Pipelines, זה אותו חשבון שירות שיצרתם בקטע הגדרת חשבון שירות עם הרשאות מפורטות. אחרת, Gemini Enterprise Agent Platform משתמשת בחשבון השירות שמוגדר כברירת מחדל ב-Compute Engine כדי להריץ צינורות עיבוד נתונים. חשבון השירות שמוגדר כברירת מחדל ב-Compute Engine נקרא כך: PROJECT_NUMBER-compute@developer.gserviceaccount.com

    3. לוחצים על הכרטיסייה Permissions.
    4. לוחצים על הענקת גישה. מופיע החלון Add principals.
    5. בתיבה New principals, מזינים את כתובת האימייל של חשבון השירות שיצרתם בשלב הקודם.
    6. ברשימה הנפתחת Role, בוחרים באפשרות Service accounts > Service account user.
    7. לוחצים על שמירה.

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

    למשל: Linux או macOS

    מחליפים את [PATH] בנתיב של קובץ ה-JSON שמכיל את המַפְתח של חשבון השירות. 

    export GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

    לדוגמה:

    export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

    למשל: Windows

    מחליפים את [PATH] בנתיב של קובץ ה-JSON שמכיל את המַפְתח של חשבון השירות, ואת [FILE_NAME] בשם הקובץ.

    באמצעות PowerShell:

    $env:GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

    לדוגמה:

    $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\[FILE_NAME].json"

    בשורת הפקודה:

    set GOOGLE_APPLICATION_CREDENTIALS=[PATH]

הגדרת תהליך העבודה באמצעות חבילת DSL של Kubeflow Pipelines

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

רכיבי צינור עיבוד נתונים ב-Kubeflow הם פונקציות של מפעל (factory functions) שיוצרות שלבים בצינור עיבוד הנתונים. כל רכיב מתאר את הקלט, הפלט וההטמעה של הרכיב. לדוגמה, בדוגמת הקוד הבאה, ds_op הוא רכיב.

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

import kfp
from google.cloud import aiplatform
from google_cloud_pipeline_components.v1.dataset import ImageDatasetCreateOp
from google_cloud_pipeline_components.v1.automl.training_job import AutoMLImageTrainingJobRunOp
from google_cloud_pipeline_components.v1.endpoint import EndpointCreateOp, ModelDeployOp

project_id = PROJECT_ID
pipeline_root_path = PIPELINE_ROOT

# Define the workflow of the pipeline.
@kfp.dsl.pipeline(
    name="automl-image-training-v2",
    pipeline_root=pipeline_root_path)
def pipeline(project_id: str):
    # The first step of your workflow is a dataset generator.
    # This step takes a Google Cloud Pipeline Component, providing the necessary
    # input arguments, and uses the Python variable `ds_op` to define its
    # output. Note that here the `ds_op` only stores the definition of the
    # output but not the actual returned object from the execution. The value
    # of the object is not accessible at the dsl.pipeline level, and can only be
    # retrieved by providing it as the input to a downstream component.
    ds_op = ImageDatasetCreateOp(
        project=project_id,
        display_name="flowers",
        gcs_source="gs://cloud-samples-data/vision/automl_classification/flowers/all_data_v2.csv",
        import_schema_uri=aiplatform.schema.dataset.ioformat.image.single_label_classification,
    )

    # The second step is a model training component. It takes the dataset
    # outputted from the first step, supplies it as an input argument to the
    # component (see `dataset=ds_op.outputs["dataset"]`), and will put its
    # outputs into `training_job_run_op`.
    training_job_run_op = AutoMLImageTrainingJobRunOp(
        project=project_id,
        display_name="train-iris-automl-mbsdk-1",
        prediction_type="classification",
        model_type="CLOUD",
        dataset=ds_op.outputs["dataset"],
        model_display_name="iris-classification-model-mbsdk",
        training_fraction_split=0.6,
        validation_fraction_split=0.2,
        test_fraction_split=0.2,
        budget_milli_node_hours=8000,
    )

    # The third and fourth step are for deploying the model.
    create_endpoint_op = EndpointCreateOp(
        project=project_id,
        display_name = "create-endpoint",
    )

    model_deploy_op = ModelDeployOp(
        model=training_job_run_op.outputs["model"],
        endpoint=create_endpoint_op.outputs['endpoint'],
        automatic_resources_min_replica_count=1,
        automatic_resources_max_replica_count=1,
    )

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

  • ‫PROJECT_ID: הפרויקט ב- Google Cloud שבו פועל צינור עיבוד הנתונים הזה.

  • ‫PIPELINE_ROOT_PATH: מציינים את ה-URI של Cloud Storage שלחשבון השירות של הצינורות יש גישה אליו. הפריטים שנוצרים מהרצות של צינורות עיבוד נתונים מאוחסנים בתיקיית השורש של צינור עיבוד הנתונים. ה-URI של Cloud Storage חייב להתחיל ב-gs://.

    אפשר להגדיר את שורש צינור עיבוד הנתונים כארגומנט של ההערה @kfp.dsl.pipeline בפונקציה של צינור עיבוד הנתונים, או כשקוראים ל-create_run_from_job_spec כדי ליצור הרצה של צינור עיבוד הנתונים.

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

אחרי שמגדירים את תהליך העבודה של צינור עיבוד הנתונים, אפשר להמשיך ולקמפל את צינור עיבוד הנתונים לפורמט YAML. קובץ ה-YAML כולל את כל המידע להפעלת הצינור ב-Agent Platform Pipelines.

from kfp import compiler

compiler.Compiler().compile(
    pipeline_func=pipeline,
    package_path='image_classif_pipeline.yaml'
)

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

אחרי שתהליך העבודה של הצינור עבר קומפילציה לפורמט YAML, אתם יכולים להשתמש בלקוח Python של Gemini Enterprise Agent Platform כדי לשלוח ולהריץ את הצינור.

import google.cloud.aiplatform as aip

# Before initializing, make sure to set the GOOGLE_APPLICATION_CREDENTIALS
# environment variable to the path of your service account.
aip.init(
    project=project_id,
    location=PROJECT_REGION,
)

# Prepare the pipeline job
job = aip.PipelineJob(
    display_name="automl-image-training-v2",
    template_path="image_classif_pipeline.yaml",
    pipeline_root=pipeline_root_path,
    parameter_values={
        'project_id': project_id
    }
)

job.submit()

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

  • ‫PROJECT_REGION: האזור שבו פועל ה-pipeline.

בדוגמה הקודמת:

  1. צינור עיבוד נתונים של Kubeflow מוגדר כפונקציית Python. הפונקציה מסומנת באמצעות ה-decorator‏ @kfp.dsl.pipeline, שמציין את השם ואת נתיב הבסיס של צינור עיבוד הנתונים. נתיב הבסיס של צינור עיבוד הנתונים הוא המיקום שבו מאוחסנים פריטי המידע שנוצרו בתהליך.
  2. השלבים בתהליך העבודה של צינור עיבוד הנתונים נוצרים באמצעות Google Cloud רכיבי צינור עיבוד הנתונים. כשמשתמשים בפלט של רכיב אחד כקלט של רכיב אחר, מגדירים את תהליך העבודה של צינור עיבוד הנתונים כתרשים. לדוגמה: training_job_run_op תלוי בפלט dataset של ds_op.
  3. מבצעים קומפילציה של צינור עיבוד הנתונים באמצעות kfp.compiler.Compiler.
  4. אתם יוצרים הרצה של צינור עיבוד נתונים ב-Agent Platform Pipelines באמצעות לקוח Python של Gemini Enterprise Agent Platform. כשמריצים צינור עיבוד נתונים, אפשר לשנות את השם ואת נתיב הבסיס שלו. אפשר לקבץ את הרצות הצינורות לפי שם הצינור. שינוי שם צינור עיבוד הנתונים יכול לעזור לכם להבחין בין הרצות של צינור עיבוד נתונים בסביבת הייצור לבין הרצות ניסיוניות.

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

בדיקת צינור עיבוד נתונים באופן מקומי (אופציונלי)

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

בדף הזה מוסבר איך להגדיר ולהריץ צינור עיבוד נתונים שמורכב משתי משימות.

הגדרת הסביבה המקומית

  1. אופציונלי: מתקינים את Docker.

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

    from kfp import dsl

# Define a component to add two numbers.
@dsl.component
def add(a: int, b: int) -> int:
    return a + b

# Define a simple pipeline using the component.
@dsl.pipeline
def addition_pipeline(x: int, y: int, z: int) -> int:
    task1 = add(a=x, b=y)
    task2 = add(a=task1.output, b=z)
    return task2.output

הפעלה של הרצה מקומית

מפעילים סשן מקומי באמצעות הפונקציה local.init(). כשמשתמשים ב-local.init(), ערכת ה-KFP SDK מריצה באופן מקומי את הצינורות והרכיבים כשקוראים להם.

כשמשתמשים ב-local.init(), צריך לציין את סוג ה-runner. סוג ה-runner מציין איך KFP צריך להריץ כל משימה.

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

from kfp import local

local.init(runner=local.DockerRunner())

pipeline_task = addition_pipeline(x=1, y=2, z=3)

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

print(f'Result: {pipeline_task.output}')

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

כדי ליצור צינור עיבוד נתונים, פועלים לפי התהליך הבא.

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

    מידע נוסף על תכנון צינורות עיבוד נתונים

  2. בונים רכיבים בהתאמה אישית שנדרשים להטמעת תהליך העבודה של למידת המכונה באמצעות Kubeflow Pipelines SDK. רכיבים הם קבוצות קוד עצמאיות שמבצעות שלב בתהליך העבודה של למידת המכונה. אפשר להשתמש באפשרויות הבאות כדי ליצור את רכיבי צינור עיבוד הנתונים.

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

      איך יוצרים רכיב של Kubeflow Pipelines

    • מטמיעים את קוד הרכיב כפונקציית Python עצמאית ומשתמשים ב-Kubeflow Pipelines SDK כדי לארוז את הפונקציה כרכיב. האפשרות הזו מקלה על פיתוח רכיבים שמבוססים על Python.

      איך יוצרים רכיבים שמבוססים על פונקציות Python

  3. יוצרים את צינור עיבוד הנתונים כפונקציית Python.

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

  4. משתמשים במהדר של Kubeflow Pipelines SDK כדי להדר את צינור עיבוד הנתונים.

    from kfp import compiler

compiler.Compiler().compile(
    pipeline_func=PIPELINE_FUNCTION,
    package_path=PIPELINE_PACKAGE_PATH)

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

    • ‫PIPELINE_FUNCTION: שם הפונקציה של צינור עיבוד הנתונים.
    • ‫PIPELINE_PACKAGE_PATH: הנתיב למיקום שבו רוצים לשמור את ה-pipeline המקומפל.

  5. מריצים את צינור עיבוד הנתונים באמצעות Google Cloud המסוף או Python.

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

אם לא תציינו חשבון שירות כשתריצו צינור עיבוד נתונים, Agent Platform Pipelines ישתמש בחשבון השירות שמוגדר כברירת מחדל ב-Compute Engine כדי להריץ את צינור עיבוד הנתונים. ב-Agent Platform Pipelines, חשבון השירות של הרצת צינור עיבוד הנתונים משמש גם כדי לאשר לצינור עיבוד הנתונים לגשת למשאבים שלGoogle Cloud . כברירת מחדל, לחשבון השירות שמוגדר כברירת מחדל ב-Compute Engine יש את התפקיד 'עריכת פרויקט'. זה עלול להעניק לצינורות עיבוד הנתונים גישה מוגזמת למשאבי Google Cloud בפרויקט Google Cloud .

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

איך יוצרים חשבון שירות ומנהלים את הגישה שניתנה לחשבון שירות באמצעות ניהול זהויות והרשאות גישה (IAM).

שמירה על עדכניות הצינורות (pipelines)

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

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