Pipeline erstellen

Mit Vertex-Pipelines können Sie Workflows für maschinelles Lernen (ML) serverlos orchestrieren. Bevor Vertex-Pipelines Ihren ML-Workflow orchestrieren können, müssen Sie Ihren Workflow als Pipeline beschreiben. ML-Pipelines sind portable und skalierbare ML-Workflows, die auf Containern und Google Cloud-Diensten basieren.

In dieser Anleitung werden die ersten Schritte zum Erstellen von ML-Pipelines beschrieben.

Welches Pipelines SDK sollte ich verwenden?

Vertex Pipelines kann Pipelines ausführen, die mit dem Kubeflow Pipelines SDK 1.6 oder höher bzw. mit TensorFlow Extended v0.30.0 oder höher erstellt wurden.

  • Wenn Sie TensorFlow in einem ML-Workflow verwenden, der Terabytes an strukturierten Daten oder Textdaten verarbeitet, sollten Sie Ihre Pipeline mit TFX erstellen.

  • Für andere Anwendungsfälle empfehlen wir, Ihre Pipeline mit dem Kubeflow Pipelines SDK zu erstellen. Wenn Sie eine Pipeline mit dem Kubeflow Pipelines SDK erstellen, können Sie Ihren Workflow implementieren, indem Sie benutzerdefinierte Komponenten erstellen oder vordefinierte Komponenten wie die Google Cloud-Pipeline-Komponenten wiederverwenden. Google Cloud-Pipeline-Komponenten erleichtern die Verwendung von Vertex-KI-Diensten wie AutoML in Ihrer Pipeline.

In dieser Anleitung wird beschrieben, wie Sie Pipelines mit dem Kubeflow Pipelines SDK erstellen.

Hinweis

Folgen Sie vor dem Erstellen und Ausführen Ihrer Pipelines der folgenden Anleitung, um Ihr Google Cloud-Projekt und Ihre Entwicklungsumgebung einzurichten.

  1. Folgen Sie der Anleitung unter Cloud-Projekt konfigurieren, um Ihr Cloud-Projekt zur Ausführung von ML-Pipelines vorzubereiten.

  2. Wenn Sie eine Pipeline mit dem Kubeflow Pipelines SDK erstellen möchten, installieren Sie das Kubeflow Pipelines SDK v1.6 oder höher.

  3. Wenn Sie Vertex AI-Dienste in Ihren Pipelines nutzen möchten, installieren Sie die Google Cloud-Pipelinekomponenten.

Einstieg in das Erstellen einer Pipeline

Um Ihren ML-Workflow in Vertex Pipelines zu orchestrieren, müssen Sie Ihren Workflow als Pipeline beschreiben. Das folgende Beispiel zeigt, wie Sie die Google Cloud-Pipelinekomponenten verwenden, um mit Vertex AI ein Dataset zu erstellen, ein Modell mit AutoML zu trainieren und das trainierte Modell für Vorhersagen bereitzustellen.

Bevor Sie das folgende Codebeispiel ausführen, müssen Sie die Authentifizierung einrichten.

import kfp
from kfp.v2 import compiler
from kfp.v2.google.client import AIPlatformClient
from google.cloud import aiplatform
from google_cloud_pipeline_components import aiplatform as gcc_aip

project_id = PROJECT_ID
region = REGION
pipeline_root_path = PIPELINE_ROOT

@kfp.dsl.pipeline(
    name="automl-image-training-v2",
    pipeline_root=pipeline_root_path)
def pipeline(project_id: str):
    ds_op = gcc_aip.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,
    )

    training_job_run_op = gcc_aip.AutoMLImageTrainingJobRunOp(
        project=project_id,
        display_name="train-iris-automl-mbsdk-1",
        prediction_type="classification",
        model_type="CLOUD",
        base_model=None,
        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,
    )

    endpoint_op = gcc_aip.ModelDeployOp(
        project=project_id, model=training_job_run_op.outputs["model"]
    )

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

api_client = AIPlatformClient(project_id=project_id, region=region)

response = api_client.create_run_from_job_spec(
    'image_classif_pipeline.json',
    pipeline_root=pipeline_root_path,
    parameter_values={
        'project_id': project_id
    })

Ersetzen Sie Folgendes:

  • PROJECT_ID: Das Google Cloud-Projekt, in dem diese Pipeline ausgeführt wird.
  • REGION: Die Region, in der diese Pipeline ausgeführt wird.
  • PIPELINE_ROOT_PATH: Geben Sie einen Cloud Storage-URI an, auf den das Pipelines-Dienstkonto zugreifen kann. Die Artefakte Ihrer Pipelineausführungen werden im Pipeline-Stammverzeichnis gespeichert.

    Das Pipeline-Stammverzeichnis kann als Argument der Annotation @kfp.dsl.pipeline für die Pipelinefunktion festgelegt werden oder es kann beim Aufrufen von create_run_from_job_spec für die Erstellung einer Pipelineausführung festgelegt werden.

Im obigen Beispiel gilt Folgendes:

  1. Eine Kubeflow-Pipeline ist als Python-Funktion definiert. Die Funktion wird mit dem Decorator @kfp.dsl.pipeline annotiert, der den Namen und den Stammpfad der Pipeline angibt. Der Stammpfad der Pipeline ist der Speicherort, an dem die Artefakte der Pipeline gespeichert sind.
  2. Die Workflow-Schritte der Pipeline werden mithilfe der Google Cloud-Pipelinekomponenten erstellt. Wenn Sie die Ausgaben einer Komponente als Eingabe für eine andere Komponente verwenden, definieren Sie den Workflow der Pipeline als Grafik. Beispiel: training_job_run_op hängt ab von der dataset-Ausgabe vonds_op.
  3. Sie kompilieren die Pipeline mit kfp.v2.compiler.Compiler.
  4. Eine Pipeline wird in Vertex-Pipelines mit kfp.v2.google.client.AIPlatformClient ausgeführt. Wenn Sie eine Pipeline ausführen, können Sie den Pipeline-Namen und den Pipeline-Stammpfad überschreiben. Pipeline-Ausführungen können mithilfe des Pipeline-Namens gruppiert werden. Durch Überschreiben des Pipeline-Namens können Sie zwischen Produktions- und experimentellen Pipeline-Ausführungen unterscheiden.

Weitere Informationen zum Erstellen von Pipelines finden Sie im Abschnitt Kubeflow-Pipelines erstellen. Sie können auch den Beispielen und Anleitungen folgen.

Kubeflow-Pipelines erstellen

Verwenden Sie den folgenden Prozess, um eine Pipeline zu erstellen.

  1. Entwerfen Sie Ihre Pipeline als eine Reihe von Komponenten. Um die Wiederverwendbarkeit zu fördern, sollte jede Komponente eine einzige Verantwortung haben. Gestalten Sie die Pipeline nach Möglichkeit immer so, dass bewährte Komponenten wie die Pipeline-Komponenten von Google Cloud wiederverwendet werden.

    Weitere Informationen zum Entwerfen von Pipelines

  2. Erstellen Sie alle benutzerdefinierten Komponenten, die zur Implementierung Ihres ML-Workflows erforderlich sind. Nutzen Sie dazu das Kubeflow Pipelines SDK Version 2. Komponenten sind eigenständige Code-Gruppen, die einen Schritt in Ihrem ML-Workflow ausführen. Verwenden Sie die folgenden Optionen, um Ihre Pipelinekomponenten zu erstellen.

    • Verpacken Sie den Code Ihrer Komponente als Container-Image. Mit dieser Option können Sie in Ihrer Pipeline Code einfügen, der in einer beliebigen Sprache geschrieben wurde, die als Container-Image verpackt werden kann.

      Kubeflow-Pipelinekomponente erstellen

    • Implementieren Sie den Code Ihrer Komponente als eigenständige Python-Funktion und verwenden Sie das Kubeflow Pipelines SDK, um Ihre Funktion als Komponente zu verpacken. Mit dieser Option lassen sich Python-basierte Komponenten einfacher erstellen.

      Funktionsbasierte Python-Komponenten erstellen

  3. Erstellen Sie Ihre Pipeline als Python-Funktion.

    Weitere Informationen zum Definieren Ihrer Pipeline als Python-Funktion

  4. Kompilieren Sie Ihre Pipeline mit dem Kubeflow Pipelines SDK v2-Compiler.

    from kfp.v2 import compiler
    
    compiler.Compiler().compile(
        pipeline_func=PIPELINE_FUNCTION,
        package_path=PIPELINE_PACKAGE_PATH)
    

    Ersetzen Sie Folgendes:

    • PIPELINE_FUNCTION: Der Name der Pipelinefunktion.
    • PIPELINE_PACKAGE_PATH: Der Pfad, unter dem Ihre kompilierte Pipeline gespeichert werden soll.
  5. Pipeline mit der Google Cloud Console oder Python ausführen

Google Cloud-Pipelinekomponenten verwenden

Die Google Cloud-Pipelinekomponenten sind Kubeflow-Pipelinekomponenten, mit denen die Verwendung von Vertex AI in Ihren Pipelines einfacher wird. Mit diesen Komponenten können Sie die folgenden Aufgaben ausführen:

  • Ein neues Dataset erstellen und Bild-, Tabellen-, Text- oder Videodaten in das Dataset laden
  • Bild-, Tabellendaten-, Text- oder Videodaten aus einem Dataset in Cloud Storage exportieren
  • AutoML verwenden, um ein Modell mit Bild-, Tabellen-, Text- oder Videodaten zu trainieren
  • Einen benutzerdefinierten Trainingsjob mit einem benutzerdefinierten Container oder einem Python-Paket ausführen
  • Ein Modell in Vertex AI hochladen
  • Einen neuen Endpunkt für Vorhersagen erstellen
  • Ein Modell für einen Endpunkt für Vorhersagen bereitstellen

Weitere Informationen zu den Google Cloud-Pipelinekomponenten

In den folgenden Beispielen erfahren Sie mehr über die Google Cloud-Pipelinekomponenten.

Maschinentyp für einen Pipelineschritt angeben

Kubeflow-Pipelinekomponenten sind Fabrikfunktionen, die Pipelineschritte erstellen. Jede Komponente beschreibt die Eingaben, Ausgaben und die Implementierung der Komponente. Im folgenden Codebeispiel ist train_op eine Komponente.

Komponenten werden verwendet, um Pipelineschritte zu erstellen. Bei der Ausführung einer Pipeline werden Schritte ausgeführt, sobald die Daten verfügbar sind, von denen sie abhängen.

Eine Trainingskomponente könnte beispielsweise eine CSV-Datei als Eingabe verwenden und zum Trainieren eines Modells verwenden. Durch Festlegen der Maschinentypparameter für den Pipelineschritt können Sie die Anforderungen jedes Schritts in Ihrer Pipeline verwalten. Wenn Sie zwei Trainingsschritte haben und in einem Schritt eine große Datendatei und im zweiten Schritt eine kleine Datendatei trainiert werden, können Sie der ersten Aufgabe mehr Speicher und CPU zuweisen und der zweiten Aufgabe weniger Ressourcen zuweisen.

Das folgende Beispiel zeigt, wie Sie die CPU-, Arbeitsspeicher- und GPU-Konfigurationseinstellungen für einen Schritt festlegen:

@dsl.pipeline(name='custom-container-pipeline')
def pipeline():
  generate = generate_op()
  train = (train_op(
      training_data=generate.outputs['training_data'],
      test_data=generate.outputs['test_data'],
      config_file=generate.outputs['config_file']).
    set_cpu_limit('CPU_LIMIT').
    set_memory_limit('MEMORY_LIMIT').
    add_node_selector_constraint(SELECTOR_CONSTRAINT).
    set_gpu_limit(GPU_LIMIT))

Ersetzen Sie Folgendes:

  • CPU_LIMIT: Das maximale CPU-Limit für diesen Operator. Dieser Stringwert kann eine Zahl (Ganzzahlwert für die Anzahl der CPUs) oder eine Zahl gefolgt von „m“ sein, was 1/1.000 bedeutet.
  • MEMORY_LIMIT: Das maximale Speicherlimit für diesen Operator. Dieser Stringwert kann eine Zahl oder eine Zahl gefolgt von „K“ (Kilobyte), „M“ (Megabyte) oder „G“ (Gigabyte) sein.
  • SELECTOR_CONSTRAINT: Jede Einschränkung ist ein Schlüssel/Wert-Paar-Label. Damit der Container auf einem Knoten ausgeführt werden kann, muss der Knoten jede der Einschränkungen als Labels haben. Beispiel:

    • 'cloud.google.com/gke-accelerator', 'nvidia-tesla-k80'

    Weitere Informationen zu GPU-Ressourcen finden Sie unter Rechenressourcen für benutzerdefiniertes Training konfigurieren.

  • GPU_LIMIT: Das GPU-Limit (positive Zahl) für den Operator.

Auf Google Cloud-Ressourcen in einer Pipeline zugreifen

Wenn Sie beim Ausführen einer Pipeline kein Dienstkonto angeben, verwendet Vertex Pipelines das Compute Engine-Standarddienstkonto zum Ausführen Ihrer Pipeline. Vertex Pipelines verwendet auch das Dienstkonto einer Pipeline-Ausführung, um Ihre Pipeline für den Zugriff auf Google Cloud-Ressourcen zu autorisieren. Das Compute Engine-Standarddienstkonto hat standardmäßig die Rolle Projektbearbeiter. Dies kann Ihren Pipelines übermäßigen Zugriff auf Google Cloud-Ressourcen in Ihrem Google Cloud-Projekt gewähren.

Es wird empfohlen, ein Dienstkonto zum Ausführen der Pipelines zu erstellen und diesem Konto detaillierte Berechtigungen für die Google Cloud-Ressourcen zu gewähren, die zum Ausführen der Pipeline erforderlich sind.

Mit Identity and Access Management ein Dienstkonto erstellen und den Zugriff steuern, der einem Dienstkonto gewährt wird

Informationen zu Ausführungs-Caching

Wenn Vertex Pipelines eine Pipeline ausführt, wird geprüft, ob in Vertex ML Metadata eine Ausführung mit der Schnittstelle jedes Pipelineschritts vorhanden ist. Die Schnittstelle des Schritts wird als Kombination aus der Pipeline, dem Pipelineschritt, den Eingaben des Pipelineschritts und den Ausgaben des Pipelineschritts definiert. Wenn in den Vertex ML Metadata eine übereinstimmende Ausführung vorhanden ist, werden die Ausgaben dieser Ausführung verwendet und der Schritt übersprungen. Dadurch können Kosten gespart werden, da Berechnungen, die bei einer vorherigen Pipelineausführung abgeschlossen wurden, übersprungen werden.

Sie können das Ausführungs-Caching für Pipelineausführungen deaktivieren, die mit Python erstellt wurden. Wenn Sie eine Pipeline mit create_run_from_job_spec ausführen, können Sie mit dem Argument enable_caching festlegen, dass diese Pipelineausführung kein Caching verwendet. Weitere Informationen zum Erstellen von Pipeline-Ausführungen

Unterschiede zwischen Vertex Pipelines und Kubeflow Pipelines

Für Entwickler, die Erfahrung mit dem Erstellen von Kubeflow-Pipelines haben, ist es wichtig zu wissen, dass sich Vertex Pipelines von Kubeflow Pipelines unterscheiden.

  • Vertex Pipelines kann Pipelines ausführen, die mit TFX v0.30.0 oder höher oder der domainspezifischen Sprache (domain-specific language, DSL) Kubeflow Pipelines SDK v2 erstellt wurden. Kubeflow Pipelines SDK v2 DSL ist im Kubeflow Pipelines SDK 1.6 oder höher verfügbar.

    Kubeflow Pipelines kann Pipelines ausführen, die mit dem Kubeflow Pipelines SDK erstellt wurden. Mit Kubeflow Pipelines v1.6 und höher können auch Pipelines ausgeführt werden, die mit dem Kubeflow Pipelines SDK v2 DSL erstellt wurden.

  • In Kubeflow Pipelines können Sie Kubernetes-Ressourcen wie Anforderungen nichtflüchtiger Volumes (Persistent Volume Claims, PVC) in Ihren Pipelines nutzen.

    In Vertex-Pipelines können Sie Google Cloud-Dienste verwenden, um Ressourcen verfügbar zu machen. So können Sie beispielsweise Cloud Storage FUSE nutzen, um auf einen Cloud Storage-Bucket als bereitgestelltes Volume in einem Pipelineschritt zuzugreifen. Wenn Ihr Cloud Storage-URI gs://example-bucket/example-pipeline ist, kann der Container Ihrer Pipelinekomponente Cloud Storage FUSE verwenden, um auf den URI als folgenden Pfad zuzugreifen: /gcs/example-bucket/example-pipeline.

  • Wenn Sie eine Pipeline mit Vertex Pipelines ausführen, muss der Stammpfad der Pipeline in der Annotation @pipeline oder beim Erstellen der Pipelineausführung angegeben worden sein.

    In Kubeflow Pipelines ist das Angeben des Pipelinestamms optional. Die Artefakte einer Pipelineausführung werden standardmäßig mit MinIO gespeichert.

  • Die folgenden Kubeflow Pipelines-Features werden in Vertex Pipelines derzeit nicht unterstützt.

    • In Kubeflow Pipelines können Sie mit Kubeflow Pipelines SDK v1 DSL festlegen, dass die Ausführung der im Cache gespeicherten Komponenten nach einer bestimmten Zeit abläuft.

      Derzeit können Sie nicht festlegen, dass Komponentenausführungen nach einer bestimmten Zeit ablaufen, wenn Sie die Kubeflow Pipelines SDK v2 DSL verwenden.

      Wenn Sie in Vertex Pipelines eine Pipeline mit create_run_from_job_spec ausführen, können Sie mit dem Argument enable_caching angeben, dass diese Pipelineausführung kein Caching verwendet.

    • In Kubeflow Pipelines können Sie Pipeline-Komponenten angeben, die rekursiv aufgerufen werden.

      Derzeit unterstützt Vertex Pipelines keine Pipeline-Komponenten, die rekursiv aufgerufen werden.

Beispiele und Anleitungen

In den folgenden Beispielen und Anleitungen finden Sie weitere Informationen zu Vertex Pipelines.

Nächste Schritte