Attivazione dell'esecuzione di una pipeline con Pub/Sub

Questa pagina mostra come scrivere, eseguire il deployment e attivare un'esecuzione della pipeline utilizzando una funzione Cloud Functions basata su eventi con un trigger Cloud Pub/Sub. Segui questi passaggi:

  1. Definisci una pipeline ML utilizzando l'SDK Kubeflow Pipelines (KFP) e compilala in un file YAML.

  2. Carica la definizione della pipeline compilata in un bucket Cloud Storage.

  3. Utilizza le funzioni Cloud Run per creare, configurare ed eseguire il deployment di una funzione attivata da un argomento Pub/Sub nuovo o esistente.

Definisci e compila una pipeline

Utilizzando l'SDK Kubeflow Pipelines, crea una pipeline pianificata e compilala in un file YAML.

Esempio di 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')

Carica il file YAML della pipeline compilata nel bucket Cloud Storage

  1. Apri il browser di Cloud Storage nella Google Cloud console.

    Browser di Cloud Storage

  2. Fai clic sul bucket Cloud Storage che hai creato quando hai configurato il progetto.

  3. Utilizzando una cartella esistente o una nuova cartella, carica il file YAML della pipeline compilata (in questo esempio hello_world_scheduled_pipeline.yaml) nella cartella selezionata.

  4. Fai clic sul file YAML caricato per accedere ai dettagli. Copia l'URI gsutil per un utilizzo successivo.

Crea una funzione Cloud Run con un trigger Pub/Sub

  1. Visita la pagina Funzioni Cloud Run nella console.

    Vai alla pagina Funzioni Cloud Run

  2. Fai clic sul pulsante Crea funzione.

  3. Nella sezione Nozioni di base, assegna un nome alla funzione (ad esempio my-scheduled-pipeline-function).

  4. Nella sezione Trigger, seleziona Cloud Pub/Sub come tipo di trigger.

    create function configuration choose pubsub as Trigger type image

  5. Nell'elenco Seleziona un argomento Cloud Pub/Sub, fai clic su Crea un argomento.

  6. Nella casella Crea un argomento , assegna un nome al nuovo argomento (ad esempio my-scheduled-pipeline-topic) e seleziona Crea argomento.

  7. Lascia tutti gli altri campi come predefiniti e fai clic su Salva per salvare la configurazione della sezione Trigger.

  8. Lascia tutti gli altri campi come predefiniti e fai clic su Avanti per procedere alla sezione Codice.

  9. In Runtime, seleziona Python 3.7.

  10. In Entry point, inserisci "subscribe" (il nome della funzione di entry point del codice di esempio).

  11. In Codice sorgente, seleziona Editor incorporato, se non è già selezionato.

  12. Nel file main.py, aggiungi il seguente codice:

      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()

    Sostituisci quanto segue:

    • PROJECT_ID: il Google Cloud progetto in cui viene eseguita questa pipeline.
    • REGION: la regione in cui viene eseguita questa pipeline.
    • PIPELINE_ROOT: specifica un URI Cloud Storage a cui può accedere l'account di servizio delle pipeline. Gli artefatti delle esecuzioni della pipeline vengono archiviati nella radice della pipeline.

  13. Nel file requirements.txt, sostituisci i contenuti con i seguenti requisiti del pacchetto:

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

  14. Fai clic su Esegui il deployment per eseguire il deployment della funzione.

Passaggi successivi