הפעלת צינור עיבוד נתונים באמצעות Pub/Sub

בדף הזה מוסבר איך לכתוב, לפרוס ולהפעיל צינור עיבוד נתונים באמצעות פונקציית Cloud מבוססת-אירועים עם טריגר Cloud Pub/Sub. איך לעשות את זה?

  1. מגדירים פייפליין של ML באמצעות Kubeflow Pipelines (KFP) SDK ומקמפלים אותו לקובץ YAML.

  2. מעלים את הגדרת הצינור המהודרת לקטגוריה של Cloud Storage.

  3. אתם יכולים להשתמש בפונקציות Cloud Run כדי ליצור, להגדיר ולפרוס פונקציה שמופעלת על ידי נושא Pub/Sub חדש או קיים.

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

באמצעות Kubeflow Pipelines SDK, יוצרים צינור עיבוד נתונים מתוזמן ומקמפלים אותו לקובץ YAML.

דוגמה hello-world-scheduled-pipeline:

from kfp import compiler
from kfp import dsl

# A simple component that prints and returns a greeting string
@dsl.component
def hello_world(message: str) -> str:
    greeting_str = f'Hello, {message}'
    print(greeting_str)
    return greeting_str

# A simple pipeline that contains a single hello_world task
@dsl.pipeline(
    name='hello-world-scheduled-pipeline')
def hello_world_scheduled_pipeline(greet_name: str):
    hello_world_task = hello_world(greet_name)

# Compile the pipeline and generate a YAML file
compiler.Compiler().compile(pipeline_func=hello_world_scheduled_pipeline,
                            package_path='hello_world_scheduled_pipeline.yaml')

העלאת קובץ YAML של פייפליין שעבר קומפילציה לקטגוריה של Cloud Storage

  1. פותחים את הדף Cloud Storage browser במסוף Google Cloud .

    Cloud Storage Browser

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

  3. מעלים את קובץ ה-YAML של צינור העיבוד (בדוגמה הזו hello_world_scheduled_pipeline.yaml) לתיקייה שנבחרה, בין אם היא קיימת או חדשה.

  4. לוחצים על קובץ ה-YAML שהועלה כדי לגשת לפרטים. מעתיקים את ה-URI של gsutil לשימוש בהמשך.

יצירת פונקציות Cloud Run עם טריגר Pub/Sub

  1. נכנסים לדף Cloud Run functions במסוף.

    כניסה לדף Cloud Run functions

  2. לוחצים על הלחצן יצירת פונקציה.

  3. בקטע Basics (עקרונות בסיסיים), נותנים שם לפונקציה (לדוגמה, my-scheduled-pipeline-function).

  4. בקטע Trigger (טריגר), בוחרים באפשרות Cloud Pub/Sub בתור סוג הטריגר.

    create function configuration choose pubsub as Trigger type image

  5. ברשימה Select a Cloud Pub/Sub topic, לוחצים על Create a topic.

  6. בתיבה יצירת נושא, נותנים שם לנושא החדש (למשל my-scheduled-pipeline-topic) ולוחצים על יצירת נושא.

  7. משאירים את כל שאר השדות כברירת מחדל ולוחצים על Save (שמירה) כדי לשמור את ההגדרה של הקטע Trigger (טריגר).

  8. משאירים את כל שאר השדות כברירת מחדל ולוחצים על הבא כדי להמשיך לקטע Code (קוד).

  9. בקטע Runtime (זמן ריצה), בוחרים באפשרות Python 3.7.

  10. בקטע Entry point (נקודת הכניסה), מזינים subscribe (שם הפונקציה של נקודת הכניסה של קוד הדוגמה).

  11. בקטע קוד מקור, בוחרים באפשרות עורך מוטבע אם היא לא מסומנת.

  12. מוסיפים את הקוד הבא לקובץ main.py:

      import base64
  import json
  from google.cloud import aiplatform

  PROJECT_ID = 'your-project-id'                     # <---CHANGE THIS
  REGION = 'your-region'                             # <---CHANGE THIS
  PIPELINE_ROOT = 'your-cloud-storage-pipeline-root' # <---CHANGE THIS

  def subscribe(event, context):
    """Triggered from a message on a Cloud Pub/Sub topic.
    Args:
          event (dict): Event payload.
          context (google.cloud.functions.Context): Metadata for the event.
    """
    # decode the event payload string
    payload_message = base64.b64decode(event['data']).decode('utf-8')
    # parse payload string into JSON object
    payload_json = json.loads(payload_message)
    # trigger pipeline run with payload
    trigger_pipeline_run(payload_json)

  def trigger_pipeline_run(payload_json):
    """Triggers a pipeline run
    Args:
          payload_json: expected in the following format:
            {
              "pipeline_spec_uri": "<path-to-your-compiled-pipeline>",
              "parameter_values": {
                "greet_name": "<any-greet-string>"
              }
            }
    """
    pipeline_spec_uri = payload_json['pipeline_spec_uri']
    parameter_values = payload_json['parameter_values']

    # Create a PipelineJob using the compiled pipeline from pipeline_spec_uri
    aiplatform.init(
        project=PROJECT_ID,
        location=REGION,
    )
    job = aiplatform.PipelineJob(
        display_name='hello-world-pipeline-cloud-function-invocation',
        template_path=pipeline_spec_uri,
        pipeline_root=PIPELINE_ROOT,
        enable_caching=False,
        parameter_values=parameter_values
    )

    # Submit the PipelineJob
    job.submit()

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

    • ‫PROJECT_ID: Google Cloud הפרויקט שבו צינור העברת הנתונים הזה פועל.
    • ‫REGION: האזור שבו הפייפליין הזה פועל.
    • ‫PIPELINE_ROOT: מציינים URI של Cloud Storage שלחשבון השירות של צינורות עיבוד הנתונים יש גישה אליו. הארטיפקטים של הפעלות צינור עיבוד הנתונים מאוחסנים בתיקיית הבסיס של צינור עיבוד הנתונים.

  13. בקובץ requirements.txt, מחליפים את התוכן בדרישות החבילה הבאות:

    google-api-python-client>=1.7.8,<2
google-cloud-aiplatform

  14. לוחצים על deploy כדי לפרוס את הפונקציה.

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