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

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

‫gcp_resource proto הוא פרמטר מיוחד שאפשר להשתמש בו ברכיב כדי לאפשר ל- Google Cloud console לספק תצוגה מותאמת אישית של היומנים והסטטוס של המשאב במסוף Gemini Enterprise Agent Platform Pipelines.

פלט הפרמטר gcp_resource

שימוש ברכיב מבוסס-מאגר

קודם כל, צריך להגדיר את הפרמטר gcp_resource ברכיב, כמו בדוגמה הבאה בקובץ component.py:

Python

במאמר התקנת Vertex AI SDK ל-Python מוסבר איך להתקין או לעדכן את Vertex AI SDK ל-Python. מידע נוסף מופיע ב מאמרי העזרה של Python API. 

# Copyright 2023 The Kubeflow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import List

from google_cloud_pipeline_components import _image
from google_cloud_pipeline_components import _placeholders
from kfp.dsl import container_component
from kfp.dsl import ContainerSpec
from kfp.dsl import OutputPath


@container_component
def dataflow_python(
    python_module_path: str,
    temp_location: str,
    gcp_resources: OutputPath(str),
    location: str = 'us-central1',
    requirements_file_path: str = '',
    args: List[str] = [],
    project: str = _placeholders.PROJECT_ID_PLACEHOLDER,
):
  # fmt: off
  """Launch a self-executing Beam Python file on Google Cloud using the
  Dataflow Runner.

  Args:
      location: Location of the Dataflow job. If not set, defaults to `'us-central1'`.
      python_module_path: The GCS path to the Python file to run.
      temp_location: A GCS path for Dataflow to stage temporary job files created during the execution of the pipeline.
      requirements_file_path: The GCS path to the pip requirements file.
      args: The list of args to pass to the Python file. Can include additional parameters for the Dataflow Runner.
      project: Project to create the Dataflow job. Defaults to the project in which the PipelineJob is run.

  Returns:
      gcp_resources: Serialized gcp_resources proto tracking the Dataflow job. For more details, see https://github.com/kubeflow/pipelines/blob/master/components/google-cloud/google_cloud_pipeline_components/proto/README.md.
  """
  # fmt: on
  return ContainerSpec(
      image=_image.GCPC_IMAGE_TAG,
      command=[
          'python3',
          '-u',
          '-m',
          'google_cloud_pipeline_components.container.v1.dataflow.dataflow_launcher',
      ],
      args=[
          '--project',
          project,
          '--location',
          location,
          '--python_module_path',
          python_module_path,
          '--temp_location',
          temp_location,
          '--requirements_file_path',
          requirements_file_path,
          '--args',
          args,
          '--gcp_resources',
          gcp_resources,
      ],
  )

בשלב הבא, בתוך מאגר התגים, מתקינים את החבילה Google Cloud Pipeline Components:

pip install --upgrade google-cloud-pipeline-components

לאחר מכן, בקוד Python, מגדירים את המשאב כפרמטר gcp_resource:

Python

במאמר התקנת Vertex AI SDK ל-Python מוסבר איך להתקין או לעדכן את Vertex AI SDK ל-Python. מידע נוסף מופיע ב מאמרי העזרה של Python API. 

from google_cloud_pipeline_components.proto.gcp_resources_pb2 import GcpResources
from google.protobuf.json_format import MessageToJson

dataflow_resources = GcpResources()
dr = dataflow_resources.resources.add()
dr.resource_type='DataflowJob'
dr.resource_uri='https://dataflow.googleapis.com/v1b3/projects/[your-project]/locations/us-east1/jobs/[dataflow-job-id]'

with open(gcp_resources, 'w') as f:
    f.write(MessageToJson(dataflow_resources))

שימוש ברכיב Python

לחלופין, אפשר להחזיר את פרמטר הפלט gcp_resources כמו כל פרמטר פלט של מחרוזת:

@dsl.component(
    base_image='python:3.9',
    packages_to_install=['google-cloud-pipeline-components==2.22.0'],
)
def launch_dataflow_component(project: str, location:str) -> NamedTuple("Outputs",  [("gcp_resources", str)]):
  # Launch the dataflow job
  dataflow_job_id = [dataflow-id]
  dataflow_resources = GcpResources()
  dr = dataflow_resources.resources.add()
  dr.resource_type='DataflowJob'
  dr.resource_uri=f'https://dataflow.googleapis.com/v1b3/projects/{project}/locations/{location}/jobs/{dataflow_job_id}'
  gcp_resources=MessageToJson(dataflow_resources)
  return gcp_resources

ערכים נתמכים של resource_type

אפשר להגדיר את resource_type כמחרוזת שרירותית, אבל רק לסוגים הבאים יש קישורים במסוף Google Cloud :

  • BatchPredictionJob
  • BigQueryJob
  • CustomJob
  • DataflowJob
  • HyperparameterTuningJob

כתיבת רכיב לביטול משאבי הבסיס

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

הביטול הוטמע בכמה Google Cloud רכיבי צינור, כולל:

  • משימת חיזוי באצווה
  • משימה ב-BigQuery ML
  • עבודות בהתאמה אישית
  • משימת אצווה של Managed Service for Apache Spark
  • משימת כוונון של היפר-פרמטרים

למידע נוסף, כולל קוד לדוגמה שמראה איך לצרף SIGTERM handler, אפשר לעיין בקישורים הבאים ב-GitHub:

כשמטמיעים את ה-handler של SIGTERM, כדאי להתייחס לנקודות הבאות:

  • הפצת הביטול פועלת רק אחרי שהרכיב פועל במשך כמה דקות. הסיבה לכך היא בדרך כלל משימות הפעלה ברקע שצריך לעבד לפני שמפעילים את הפונקציות לטיפול באותות של Python.
  • יכול להיות שבחלק Google Cloud מהמשאבים לא מיושם ביטול. לדוגמה, יצירה או מחיקה של נקודת קצה או מודל ב-Gemini Enterprise Agent Platform יכולה ליצור פעולה ממושכת שמקבלת בקשת ביטול דרך ה-API בארכיטקטורת REST שלה, אבל לא מבצעת את פעולת הביטול עצמה.