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

אפשר להפעיל מחדש משימות ספציפיות מהפעלות של צינורות מכונת למידה (ML) שהושלמו, בוטלו או נכשלו. כשמפעילים מחדש ריצה, אפשר לשנות את ההגדרות ברמת המשימה או לבחור לדלג על משימות, ואז ליצור ריצה על סמך ההגדרה המעודכנת. הפעלת צינור העיבוד החדש שומרת הפניה להפעלת צינור העיבוד המקורית לצורך מעקב. אם משימה מסוימת הצליחה בהרצה קודמת, Gemini Enterprise Agent Platform Pipelines משתמש מחדש בתוצאות שנשמרו במטמון של אותה משימה. אחרת, אם השלב נכשל, צינורות של Agent Platform מריצים את השלב במהלך ההפעלה מחדש של הצינור.

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

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

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

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

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

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

כדי להפעיל מחדש צינור עיבוד נתונים, משתמשים ב-Agent Platform SDK ל-Python.

Python

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

from google.cloud import aiplatform
from google.cloud.aiplatform.preview.pipelinejob.pipeline_jobs import (
  _PipelineJob as PipelineJob
)
from google.cloud.aiplatform_v1beta1.types.ui_pipeline_spec import RuntimeArtifact
from google.protobuf.struct_pb2 import Value
from google.cloud.aiplatform_v1beta1.types import PipelineTaskRerunConfig
aiplatform.init(project="PROJECT_ID", location="LOCATION")
job = aiplatform.PipelineJob.get(resource_name="PIPELINE_RUN_RESOURCE_NAME")
original_job_name = job.resource_name
rerun_task_id = None
skip_failed_task_id = None
task_inputs_override = PipelineTaskRerunConfig.Inputs(
    parameter_values={
      "TASK_PARAMETER_1": Value(TASK_PARAMETER_1_VALUE),
      "TASK_PARAMETER_2": Value(TASK_PARAMETER_2_VALUE)
    }
)
for task in job.task_details:
    if task.task_name == RERUN_TASK_NAME:
        rerun_task_id = task.task_id
    if task.task_name == SKIP_FAILED_TASK_NAME:
        skip_failed_task_id = task.task_id
pipeline_job.rerun(original_pipelinejob_name=original_job_name,
  pipeline_task_rerun_configs=[
    PipelineTaskRerunConfig(task_id = rerun_task_id,
      skip_task = False,
      inputs = PipelineTaskRerunConfig.Inputs(task_inputs_override)
    ),
    PipelineTaskRerunConfig(task_id = skip_failed_task_id,
      skip_task = True
    )
  ],
  parameter_values=PIPELINE_PARAMETER_VALUES,
  job_id=RERUN_PIPELINE_JOB_ID)

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

  • ‫PROJECT_ID: Google Cloud הפרויקט שמכיל את ההרצה של צינור הנתונים.
  • ‫LOCATION: האזור שבו נמצאת ההרצה של צינור העיבוד. מידע נוסף על האזורים שבהם זמין Agent Platform Pipelines מופיע במדריך למיקומים של Agent Platform של Gemini Enterprise.

  • ‫PIPELINE_RUN_RESOURCE_NAME: שם המשאב המוגדר במלואו של הפעלת צינור שרוצים להפעיל מחדש, שהסתיימה, נכשלה או בוטלה. מזינים את שם המשאב בפורמט projects/PROJECT_NUMBER/locations/LOCATION/pipelineJobs/PIPELINE_RUN_ID, כאשר:

    • ‫PROJECT_NUMBER: מספר הפרויקט. אפשר למצוא את מספר הפרויקט במסוף Google Cloud . מידע נוסף זמין במאמר איך מוצאים את השם, המספר והמזהה של הפרויקט.
    • ‫PIPELINE_RUN_ID עם המזהה הייחודי של הפעלת צינור עיבוד הנתונים שרוצים להפעיל מחדש. המזהה מוצג בכרטיסייה Runs בדף Pipelines במסוף Google Cloud .
  • ‫RERUN_TASK_NAME: השם של המשימה שרוצים להפעיל מחדש עם פרמטרים מעודכנים.
  • ‫SKIP_FAILED_TASK_NAME: השם של המשימה שנכשלה ורוצים לדלג עליה בהפעלה מחדש.
  • ‫TASK_PARAMETER_1 ו-TASK_PARAMETER_2: השמות של הפרמטרים של המשימה שרוצים לשנות בהרצה חוזרת של צינור העיבוד.
  • ‫TASK_PARAMETER_1_VALUE ו-TASK_PARAMETER_2_VALUE: הערכים החדשים של TASK_PARAMETER_1 ו-TASK_PARAMETER_2 בהתאמה בהרצה חוזרת של הצינור.
  • ‫PIPELINE_PARAMETER_VALUES: אופציונלי. ערכי הפרמטרים המעודכנים ברמת ההרצה של צינור עיבוד הנתונים שבהם יש להשתמש בהרצה מחדש של צינור עיבוד הנתונים.
  • ‫RERUN_PIPELINE_JOB_ID: אופציונלי. מזהה ייחודי להקצאה למשימה החדשה של הפעלה מחדש של הפייפליין.