Klassische Vorlagen ausführen

Nachdem Sie die Dataflow-Vorlage erstellt und bereitgestellt haben, führen Sie die Vorlage über die Google Cloud Console, die REST API oder das gcloud-Befehlszeilentool aus. Dataflow-Vorlagenjobs können in zahlreichen Umgebungen bereitgestellt werden, unter anderem in der App Engine-Standardumgebung, in Cloud Functions und in anderen eingeschränkten Umgebungen.

Hinweis: Neben der Vorlagendatei richtet sich die Ausführung der Pipeline nach Vorlage außerdem nach Dateien, die während der Vorlagenerstellung bereitgestellt worden sind und auf die zu diesem Zeitpunkt verwiesen wurde. Wenn Sie die bereitgestellten Dateien verschieben oder entfernen, schlägt der Pipelinejob fehl.

Mit der Cloud Console

Sie können die Cloud Console verwenden, um von Google bereitgestellte und benutzerdefinierte Dataflow-Vorlagen auszuführen.

Von Google bereitgestellte Vorlagen

Eine von Google bereitgestellte Vorlage ausführen:

  1. Rufen Sie in der Cloud Console die Seite "Dataflow" auf.
    2. Zur Seite "Dataflow"
  2. Klicken Sie auf JOB AUS VORLAGE ERSTELLEN.
    3. Schaltfläche
  3. Wählen Sie im Drop-down-Menü Dataflow-Vorlage die von Google bereitgestellte Vorlage aus, die Sie ausführen möchten.
    4. Formular für die WordCount-Vorlagenausführung
  4. Geben Sie einen Jobnamen in das Feld Jobname ein. Der Jobname muss dem regulären Ausdruck [a-z]([-a-z0-9]{0,38}[a-z0-9])? entsprechen, um gültig zu sein.
  5. Geben Sie Ihre Parameterwerte in die verfügbaren Parameterfelder ein. Den Abschnitt Zusätzliche Parameter sollten Sie nicht benötigen, wenn Sie eine von Google bereitgestellte Vorlage verwenden.
  6. Klicken Sie auf Job ausführen.

Benutzerdefinierte Vorlagen

Benutzerdefinierte Vorlage ausführen:

  1. Rufen Sie in der Cloud Console die Seite "Dataflow" auf.
    2. Zur Seite "Dataflow"
  2. Klicken Sie auf JOB AUS VORLAGE ERSTELLEN.
    3. Schaltfläche
  3. Wählen Sie im Drop-down-Menü Dataflow-Vorlage die Option Benutzerdefinierte Vorlage aus.
    4. Formular für die Ausführung benutzerdefinierter Vorlagen
  4. Geben Sie einen Jobnamen in das Feld Jobname ein. Der Jobname muss dem regulären Ausdruck [a-z]([-a-z0-9]{0,38}[a-z0-9])? entsprechen, um gültig zu sein.
  5. Geben Sie den Cloud Storage-Pfad zu Ihrer Vorlage in das Vorlagenfeld für den Cloud Storage-Pfad ein.
  6. Wenn für Ihre Vorlage Parameter erforderlich sind, klicken Sie im Abschnitt Zusätzliche Parameter auf Zeile hinzufügen. Geben Sie den Namen und den Wert des Parameters ein. Wiederholen Sie diesen Schritt für jeden benötigten Parameter.
  7. Klicken Sie auf Job ausführen.

REST API verwenden

Senden Sie zum Ausführen einer Vorlage mit einer REST API-Anfrage eine HTTP-POST-Anfrage mit Ihrer Projekt-ID. Diese Anfrage erfordert eine Autorisierung.

Weitere Informationen zu den verfügbaren Parametern finden Sie in der REST API-Referenz für projects.templates.launch.

Beispiel 1: Batchjob aus benutzerdefinierter Vorlage erstellen

Durch die Beispielanfrage projects.templates.launch wird aus einer Vorlage ein Batch-Job erstellt, der eine Textdatei liest und eine Ausgabetextdatei schreibt. Bei Erfolg enthält der Antworttext eine Instanz von LaunchTemplateResponse.

Die folgenden Werte müssen geändert werden:

  • Ersetzen Sie YOUR_PROJECT_ID durch Ihre Projekt-ID.
  • Ersetzen Sie JOB_NAME durch einen Jobnamen Ihrer Wahl. Der Jobname muss dem regulären Ausdruck [a-z]([-a-z0-9]{0,38}[a-z0-9])? entsprechen, um gültig zu sein.
  • Ersetzen Sie YOUR_BUCKET_NAME durch den Namen des Cloud Storage-Buckets.
  • Legen Sie für gcsPath den Cloud Storage-Speicherort der Vorlagendatei fest.
  • Stellen Sie das Flag parameters auf Ihre Liste der Schlüssel/Wert-Paare ein.
  • Legen Sie für tempLocation einen Speicherort fest, für den Sie die Schreibberechtigung haben. Dieser Wert ist erforderlich, um von Google bereitgestellte Vorlagen auszuführen.
    POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://YOUR_BUCKET_NAME/templates/TemplateName
    {
        "jobName": "JOB_NAME",
        "parameters": {
            "inputFile" : "gs://YOUR_BUCKET_NAME/input/my_input.txt",
            "outputFile": "gs://YOUR_BUCKET_NAME/output/my_output"
        },
        "environment": {
            "tempLocation": "gs://YOUR_BUCKET_NAME/temp",
            "zone": "us-central1-f"
        }
    }

Beispiel 2: Streamingjob aus benutzerdefinierter Vorlage erstellen

Durch die Beispielanfrage projects.templates.launch wird ein Streaming-Job aus einer Vorlage erstellt, die aus einem Pub/Sub-Thema liest und in eine BigQuery-Tabelle schreibt. Die BigQuery-Tabelle muss bereits mit einem geeigneten Schema vorhanden sein. Bei Erfolg enthält der Antworttext eine Instanz von LaunchTemplateResponse.

Die folgenden Werte müssen geändert werden:

  • Ersetzen Sie YOUR_PROJECT_ID durch Ihre Projekt-ID.
  • Ersetzen Sie JOB_NAME durch einen Jobnamen Ihrer Wahl. Der Jobname muss dem regulären Ausdruck [a-z]([-a-z0-9]{0,38}[a-z0-9])? entsprechen, um gültig zu sein.
  • Ersetzen Sie YOUR_BUCKET_NAME durch den Namen Ihres Cloud Storage-Buckets.
  • Ersetzen Sie YOUR_TOPIC_NAME durch Ihren Pub/Sub-Themennamen.
  • Ersetzen Sie YOUR_DATASET durch Ihr BigQuery-Dataset und ersetzen Sie YOUR_TABLE_NAME durch Ihren BigQuery-Tabellennamen.
  • Legen Sie für gcsPath den Cloud Storage-Speicherort der Vorlagendatei fest.
  • Stellen Sie das Flag parameters auf Ihre Liste der Schlüssel/Wert-Paare ein.
  • Legen Sie für tempLocation einen Speicherort fest, für den Sie die Schreibberechtigung haben. Dieser Wert ist erforderlich, um von Google bereitgestellte Vorlagen auszuführen.
    POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://YOUR_BUCKET_NAME/templates/TemplateName
    {
        "jobName": "JOB_NAME",
        "parameters": {
            "topic": "projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME",
            "table": "YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME"
        },
        "environment": {
            "tempLocation": "gs://YOUR_BUCKET_NAME/temp",
            "zone": "us-central1-f"
        }
    }

Beispiel 3: Aus benutzerdefinierter Vorlage erstellten Streamingjob aktualisieren

Diese Beispielanfrage projects.templates.launch zeigt Ihnen, wie Sie eine Vorlage für einen Streamingjob aktualisieren.

  1. Führen Sie Beispiel 2: Streamingjob aus benutzerdefinierter Vorlage erstellen aus, um einen Streaming-Vorlagenjob zu starten.
  2. Senden Sie die folgende HTTP-POST-Anfrage mit den folgenden geänderten Werten:
    • Ersetzen Sie YOUR_PROJECT_ID durch Ihre Projekt-ID.
    • Ersetzen Sie JOB_NAME durch einen Jobnamen Ihrer Wahl. Der Jobname muss dem regulären Ausdruck [a-z]([-a-z0-9]{0,38}[a-z0-9])? entsprechen, um gültig zu sein.
    • Ersetzen Sie YOUR_BUCKET_NAME durch den Namen Ihres Cloud Storage-Buckets.
    • Ersetzen Sie YOUR_TOPIC_NAME durch Ihren Pub/Sub-Themennamen.
    • Ersetzen Sie YOUR_DATASET durch Ihr BigQuery-Dataset und ersetzen Sie YOUR_TABLE_NAME durch Ihren BigQuery-Tabellennamen.
    • Legen Sie für gcsPath den Cloud Storage-Speicherort der Vorlagendatei fest.
    • Stellen Sie das Flag parameters auf Ihre Liste der Schlüssel/Wert-Paare ein.
    • Legen Sie für tempLocation einen Speicherort fest, für den Sie die Schreibberechtigung haben. Dieser Wert ist erforderlich, um von Google bereitgestellte Vorlagen auszuführen.
        POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://YOUR_BUCKET_NAME/templates/TemplateName
    {
        "jobName": "JOB_NAME",
        "parameters": {
            "topic": "projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME",
            "table": "YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME"
        },
        "environment": {
            "tempLocation": "gs://YOUR_BUCKET_NAME/temp",
            "zone": "us-central1-f"
        }
        "update": true
    }
  3. Rufen Sie die Dataflow-Monitoring-Oberfläche auf und prüfen Sie, ob ein neuer Job mit demselben Namen erstellt wurde. Dieser Job hat den Status Aktualisiert.

Google API-Clientbibliotheken verwenden

Mit Google API-Clientbibliotheken können Sie auf einfache Weise Dataflow REST APIs aufrufen. In diesem Beispielskript wird die Google API-Clientbibliothek für Python verwendet.

Geben Sie für dieses Beispiel die folgenden Variablen an:

  • project: Auf Ihre Projekt-ID festlegen.
  • job: Auf einen eindeutigen Jobnamen Ihrer Wahl festlegen. Der Jobname muss dem regulären Ausdruck [a-z]([-a-z0-9]{0,38}[a-z0-9])? entsprechen, um gültig zu sein.
  • template: Auf den Cloud Storage-Speicherort der Vorlagendatei festlegen.
  • parameters: Ein Wörterbuch mit den Vorlagenparametern festlegen.
dataflow/run_template/main.py 
from googleapiclient.discovery import build

# project = 'your-gcp-project'
# job = 'unique-job-name'
# template = 'gs://dataflow-templates/latest/Word_Count'
# parameters = {
#     'inputFile': 'gs://dataflow-samples/shakespeare/kinglear.txt',
#     'output': 'gs://<your-gcs-bucket>/wordcount/outputs',
# }

dataflow = build('dataflow', 'v1b3')
request = dataflow.projects().templates().launch(
    projectId=project,
    gcsPath=template,
    body={
        'jobName': job,
        'parameters': parameters,
    }
)

response = request.execute()

Weitere Informationen zu den verfügbaren Optionen finden Sie in der Referenz zur Cloud Dataflow REST API unter der Methode projects.templates.launch.

gcloud verwenden

Hinweis: Wenn Sie Vorlagen mit dem gcloud-Befehlszeilentool ausführen möchten, benötigen Sie Cloud SDK 138.0.0 oder eine höhere Version.

Mit dem gcloud-Befehlszeilentool kann eine benutzerdefinierte oder eine von Google bereitgestellte Vorlage mit dem Befehl gcloud dataflow jobs run ausgeführt werden. Beispiele für die Ausführung von Vorlagen, die von Google bereitgestellt wurden, sind auf der Seite Von Google bereitgestellte Vorlagen dokumentiert.

Legen Sie für die folgenden Beispiele zu benutzerdefinierten Vorlagen die Werte so fest:

  • Ersetzen Sie JOB_NAME durch einen Jobnamen Ihrer Wahl. Der Jobname muss dem regulären Ausdruck [a-z]([-a-z0-9]{0,38}[a-z0-9])? entsprechen, um gültig zu sein.
  • Ersetzen Sie YOUR_BUCKET_NAME durch den Namen Ihres Cloud Storage-Buckets.
  • Sie müssen das Flag --gcs-location einfügen. Legen Sie für --gcs-location den Cloud Storage-Speicherort der Vorlagendatei fest.
  • Legen Sie für --parameters die durch Kommas abgetrennte Liste der Parameter fest, die an den Job übergeben werden sollen. Leerzeichen zwischen Kommas und Werten sind nicht zugelassen.

Beispiel 1: Benutzerdefinierte Vorlage, Batchjob

In diesem Beispiel wird aus einer Vorlage ein Batchjob erstellt, der eine Textdatei einliest und eine Ausgabetextdatei schreibt.

    gcloud dataflow jobs run JOB_NAME \
        --gcs-location gs://YOUR_BUCKET_NAME/templates/MyTemplate \
        --parameters inputFile=gs://YOUR_BUCKET_NAME/input/my_input.txt,outputFile=gs://YOUR_BUCKET_NAME/output/my_output

Die Anfrage gibt eine Antwort im folgenden Format zurück.

    id: 2016-10-11_17_10_59-1234530157620696789
    projectId: YOUR_PROJECT_ID
    type: JOB_TYPE_BATCH

Beispiel 2: Benutzerdefinierte Vorlage, Streamingjob

In diesem Beispiel wird aus einer Vorlage ein Streamingjob erstellt, der aus einem Pub/Sub-Thema liest und in eine BigQuery-Tabelle schreibt. Die BigQuery-Tabelle muss bereits mit einem geeigneten Schema vorhanden sein.

    gcloud dataflow jobs run JOB_NAME \
        --gcs-location gs://YOUR_BUCKET_NAME/templates/MyTemplate \
        --parameters topic=projects/project-identifier/topics/resource-name,table=my_project:my_dataset.my_table_name

Die Anfrage gibt eine Antwort im folgenden Format zurück.

    id: 2016-10-11_17_10_59-1234530157620696789
    projectId: YOUR_PROJECT_ID
    type: JOB_TYPE_STREAMING

Eine vollständige Liste der Flags für den Befehl gcloud dataflow jobs run finden Sie in der Referenz für das gcloud-Tool.

Überwachung und Fehlerbehebung

Mit der Dataflow-Monitoring-Oberfläche können Sie Ihre Dataflow-Jobs überwachen. Schlägt ein Job fehl, finden Sie im Leitfaden Pipelinefehler beheben Tipps und Strategien zur Fehlerbehebung sowie einen Katalog mit häufig auftretenden Fehlern.