Dataflow-Pipeline mit Go erstellen

Dataflow-Pipeline mit Go erstellen

Auf dieser Seite erfahren Sie, wie Sie mit dem Apache Beam SDK für Go ein Programm erstellen, das eine Pipeline definiert. Anschließend führen Sie die Pipeline lokal und im Dataflow-Dienst aus. Eine Einführung in die WordCount-Pipeline finden Sie im Video How to use WordCount in Apache Beam.

Hinweis

1. Melden Sie sich bei Ihrem Google Cloud-Konto an. Wenn Sie mit Google Cloud noch nicht vertraut sind, erstellen Sie ein Konto, um die Leistungsfähigkeit unserer Produkte in der Praxis sehen und bewerten zu können. Neukunden erhalten außerdem ein Guthaben von 300 $, um Arbeitslasten auszuführen, zu testen und bereitzustellen.
2. Installieren Sie die Google Cloud CLI.

3. Führen Sie folgenden Befehl aus, um die gcloud CLI zu initialisieren:

   gcloud init

4. Google Cloud-Projekt erstellen oder auswählen.

   Hinweis: Wenn Sie die Ressourcen, die Sie in diesem Verfahren erstellen, nicht behalten möchten, erstellen Sie ein Projekt, anstatt ein vorhandenes Projekt auszuwählen. Wenn Sie fertig sind, können Sie das Projekt löschen und dadurch alle mit dem Projekt verknüpften Ressourcen entfernen.

   • Erstellen Sie ein Google Cloud-Projekt:

     gcloud projects create PROJECT_ID

     Ersetzen Sie PROJECT_ID durch einen Namen für das Google Cloud-Projekt, das Sie erstellen.

   • Wählen Sie das von Ihnen erstellte Google Cloud-Projekt aus:

     gcloud config set project PROJECT_ID

     Ersetzen Sie PROJECT_ID durch den Namen Ihres Google Cloud-Projekts.

5. Die Abrechnung für das Google Cloud-Projekt muss aktiviert sein.

6. Aktivieren Sie die Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, and Cloud Resource Manager APIs:

   gcloud services enable dataflow compute_component logging storage_component storage_api cloudresourcemanager.googleapis.com

7. Erstellen Sie lokale Anmeldedaten zur Authentifizierung für Ihr Google-Konto:

   gcloud auth application-default login

8. Gewähren Sie Ihrem Google-Konto Rollen. Führen Sie den folgenden Befehl für jede der folgenden IAM-Rollen einmal aus: roles/iam.serviceAccountUser

   gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE

   • Ersetzen Sie PROJECT_ID durch Ihre Projekt-ID.
   • Ersetzen Sie EMAIL_ADDRESS durch Ihre E-Mail-Adresse.
   • Ersetzen Sie ROLE durch jede einzelne Rolle.
9. Installieren Sie die Google Cloud CLI.

10. Führen Sie folgenden Befehl aus, um die gcloud CLI zu initialisieren:

    gcloud init

11. Google Cloud-Projekt erstellen oder auswählen.

    Hinweis: Wenn Sie die Ressourcen, die Sie in diesem Verfahren erstellen, nicht behalten möchten, erstellen Sie ein Projekt, anstatt ein vorhandenes Projekt auszuwählen. Wenn Sie fertig sind, können Sie das Projekt löschen und dadurch alle mit dem Projekt verknüpften Ressourcen entfernen.

    • Erstellen Sie ein Google Cloud-Projekt:

      gcloud projects create PROJECT_ID

      Ersetzen Sie PROJECT_ID durch einen Namen für das Google Cloud-Projekt, das Sie erstellen.

    • Wählen Sie das von Ihnen erstellte Google Cloud-Projekt aus:

      gcloud config set project PROJECT_ID

      Ersetzen Sie PROJECT_ID durch den Namen Ihres Google Cloud-Projekts.

12. Die Abrechnung für das Google Cloud-Projekt muss aktiviert sein.

13. Aktivieren Sie die Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, and Cloud Resource Manager APIs:

    gcloud services enable dataflow compute_component logging storage_component storage_api cloudresourcemanager.googleapis.com

14. Erstellen Sie lokale Anmeldedaten zur Authentifizierung für Ihr Google-Konto:

    gcloud auth application-default login

15. Gewähren Sie Ihrem Google-Konto Rollen. Führen Sie den folgenden Befehl für jede der folgenden IAM-Rollen einmal aus: roles/iam.serviceAccountUser

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE

    • Ersetzen Sie PROJECT_ID durch Ihre Projekt-ID.
    • Ersetzen Sie EMAIL_ADDRESS durch Ihre E-Mail-Adresse.
    • Ersetzen Sie ROLE durch jede einzelne Rolle.

16. Weisen Sie Ihrem Compute Engine-Standarddienstkonto Rollen zu. Führen Sie den folgenden Befehl für jede der folgenden IAM-Rollen einmal aus:

    • roles/dataflow.admin
    • roles/dataflow.worker
    • roles/storage.objectAdmin

    gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:[e-mail]" --role=SERVICE_ACCOUNT_ROLE

    • Ersetzen Sie PROJECT_ID durch Ihre Projekt-ID.
    • Ersetzen Sie PROJECT_NUMBER durch die Projekt-ID. Ihre Projektnummer finden Sie unter Projekte identifizieren oder verwenden Sie den Befehl gcloud projects describe.
    • Ersetzen Sie SERVICE_ACCOUNT_ROLE durch jede einzelne Rolle.
17. Erstellen Sie einen Cloud Storage-Bucket und konfigurieren Sie ihn so:
    • Legen Sie die Speicherklasse auf S (Standard).
    • Legen Sie als Speicherort Folgendes fest: US (USA)
    • Ersetzen Sie BUCKET_NAME durch einen eindeutigen Bucket-Namen. Der Bucket-Name darf keine vertraulichen Informationen enthalten, da der Bucket-Namespace global und öffentlich sichtbar ist.

    gcloud storage buckets create gs://BUCKET_NAME --default-storage-class STANDARD --location US

18. Kopieren Sie die Google Cloud-Projekt-ID und den Namen des Cloud Storage-Buckets. Sie benötigen diese Werte später in dieser Kurzanleitung.

Entwicklungsumgebung einrichten

Das Apache Beam SDK ist ein Open-Source-Programmiermodell für Datenpipelines. Sie definieren eine Pipeline mit einem Apache Beam-Programm und wählen dann einen Runner wie Dataflow aus, um Ihre Pipeline auszuführen.

Wir empfehlen die Verwendung der neuesten Version von Go, wenn Sie mit dem Apache Beam SDK für Go arbeiten. Wenn Sie nicht die neueste Version von Go installiert haben, können Sie Go im Leitfaden zum Herunterladen und Installieren von Go für Ihr Betriebssystem verwenden.

Führen Sie den folgenden Befehl in Ihrem lokalen Terminal aus, um die installierte Version von Go zu prüfen:

go version

Beispiel für Beam-Wordcount ausführen

Das Apache Beam SDK für Go enthält ein wordcount-Pipeline-Beispiel. Das wordcount-Beispiel führt Folgendes aus:

1. Liest eine Textdatei als Eingabe. Standardmäßig wird eine Textdatei in einem Cloud Storage-Bucket mit dem Ressourcennamen gs://dataflow-samples/shakespeare/kinglear.txt gelesen.
2. Sie parst jede Zeile und unterteilt sie in Wörter.
3. Sie misst die Häufigkeit der tokenisierten Wörter.

Verwenden Sie den folgenden Befehl, um die neueste Version des Beam-wordcount-Beispiels auf Ihrem lokalen Computer auszuführen. Das Flag input gibt die zu lesende Datei an und das Flag output gibt den Dateinamen für die Ausgabe der Häufigkeit an.

go run github.com/apache/beam/sdks/v2/go/examples/wordcount@latest \
  --input gs://dataflow-samples/shakespeare/kinglear.txt \
  --output outputs

Sehen Sie sich nach Abschluss der Pipeline die Ausgabeergebnisse an:

more outputs*

Drücken Sie zum Beenden q.

Pipelinecode ändern

Die Beam-wordcount-Pipeline unterscheidet zwischen Groß- und Kleinbuchstaben. Die folgenden Schritte zeigen, wie Sie Ihr eigenes Go-Modul erstellen, die wordcount-Pipeline so ändern, dass sie Groß- und Kleinschreibung nicht beachtet, und sie auf Dataflow ausführen.

Go-Modul erstellen

Gehen Sie so vor, um Änderungen am Pipelinecode vorzunehmen.

1. Erstellen Sie ein Verzeichnis für Ihr Go-Modul an einem Speicherort Ihrer Wahl:

   mkdir wordcount
   cd wordcount
   

2. Go-Modul erstellen Verwenden Sie für dieses Beispiel example/dataflow als Modulpfad.

   go mod init example/dataflow
   

3. Laden Sie die neueste Version des Codes wordcount aus dem Apache Beam GitHub-Repository herunter. Legen Sie diese Datei in dem Verzeichnis wordcount ab, das Sie erstellt haben.

4. Wenn Sie ein Nicht-Linux-Betriebssystem verwenden, müssen Sie das Go-Paket unix abrufen. Dieses Paket ist erforderlich, um Pipelines im Dataflow-Dienst auszuführen.

   go get -u golang.org/x/sys/unix
   

5. Prüfen Sie, ob die Datei go.mod mit dem Quellcode des Moduls übereinstimmt:

   go mod tidy
   

Unveränderte Pipeline ausführen

Prüfen Sie, ob die unveränderte wordcount-Pipeline lokal ausgeführt wird.

1. Erstellen Sie die Pipeline über das Terminal und führen Sie sie lokal aus:

    go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
        --output outputs
   

2. Ausgabeergebnisse ansehen:

    more outputs*
   

3. Drücken Sie zum Beenden q.

Pipelinecode ändern

Wenn Sie die Pipeline so ändern möchten, dass sie nicht mehr zwischen Groß- und Kleinschreibung unterscheidet, ändern Sie den Code so, dass die Funktion strings.ToLower auf alle Wörter angewendet wird.

1. Öffnen Sie die Datei wordcount.go in einem Editor Ihrer Wahl.

2. Untersuchen Sie den Block init (Kommentare wurden zur besseren Übersichtlichkeit entfernt):

    func init() {
      register.DoFn3x0[context.Context, string, func(string)](&extractFn{})
      register.Function2x1(formatFn)
      register.Emitter1[string]()
    }
   

3. Fügen Sie eine neue Zeile hinzu, um die Funktion strings.ToLower zu registrieren:

    func init() {
      register.DoFn3x0[context.Context, string, func(string)](&extractFn{})
      register.Function2x1(formatFn)
      register.Emitter1[string]()
      register.Function1x1(strings.ToLower)
    }
   

4. Sehen Sie sich die Funktion CountWords an:

    func CountWords(s beam.Scope, lines beam.PCollection) beam.PCollection {
      s = s.Scope("CountWords")
   
      // Convert lines of text into individual words.
      col := beam.ParDo(s, &extractFn{SmallWordLength: *smallWordLength}, lines)
   
      // Count the number of times each word occurs.
      return stats.Count(s, col)
    }
   

5. Fügen Sie zum Entfernen der Kleinbuchstaben ein ParDo hinzu, das strings.ToLower auf jedes Wort anwendet:

    func CountWords(s beam.Scope, lines beam.PCollection) beam.PCollection {
      s = s.Scope("CountWords")
   
      // Convert lines of text into individual words.
      col := beam.ParDo(s, &extractFn{SmallWordLength: *smallWordLength}, lines)
   
      // Map all letters to lowercase.
      lowercaseWords := beam.ParDo(s, strings.ToLower, col)
   
      // Count the number of times each word occurs.
      return stats.Count(s, lowercaseWords)
    }
   

6. Speichern Sie die Datei.

Aktualisierte Pipeline lokal ausführen

Führen Sie die aktualisierte wordcount-Pipeline lokal aus und prüfen Sie, ob sich die Ausgabe geändert hat.

1. Erstellen Sie die geänderte wordcount-Pipeline und führen Sie sie aus:

    go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
        --output outputs
   

2. Sehen Sie sich die Ausgabeergebnisse der geänderten Pipeline an. Alle Wörter sollten Kleinbuchstaben sein.

    more outputs*
   

3. Drücken Sie zum Beenden q.

Pipeline im Dataflow-Dienst ausführen

Führen Sie den folgenden Befehl aus, um das aktualisierte wordcount-Beispiel im Dataflow-Dienst auszuführen:

go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
    --output gs://BUCKET_NAME/results/outputs \
    --runner dataflow \
    --project PROJECT_ID \
    --region DATAFLOW_REGION \
    --staging_location gs://BUCKET_NAME/binaries/

Dabei gilt:

• BUCKET_NAME: der Name des Cloud Storage-Buckets.

• PROJECT_ID: die Google Cloud-Projekt-ID.

• DATAFLOW_REGION: die Region, in der Sie den Dataflow-Job bereitstellen möchten. Beispiel: europe-west1 Eine Liste der verfügbaren Dataflow-Standorte finden Sie hier. Das Flag --region überschreibt die Standardregion, die auf dem Metadatenserver, auf Ihrem lokalen Client oder in Umgebungsvariablen festgelegt ist.

Hinweis: Um ein nutzerverwaltetes Worker-Dienstkonto anzugeben, schließen Sie die --service_account_email Pipelineoption mit ein. Nutzerverwaltete Worker-Dienstkonten werden für Produktionsarbeitslasten empfohlen. Wenn Sie beim Erstellen eines Jobs kein Worker-Dienstkonto angeben, verwendet Dataflow das Compute Engine-Standarddienstkonto.

Ergebnisse ansehen

Sie können eine Liste Ihrer Dataflow-Jobs in der Google Cloud Console aufrufen. Rufen Sie in der Google Cloud Console die Dataflow-Seite Jobs auf.

ZU JOBS

Auf der Seite Jobs werden Details zum wordcount-Job angezeigt, z. B. der Status Aktiv und dann Erfolgreich.

Wenn Sie eine Pipeline mit Dataflow ausführen, werden die Ergebnisse in einem Cloud Storage-Bucket gespeichert. Sehen Sie sich die Ausgabeergebnisse entweder mit der Google Cloud Console oder mit dem lokalen Terminal an.

Console

Wechseln Sie zur Seite Cloud Storage-Buckets, um Ihre Ergebnisse in der Google Cloud Console anzusehen.

Buckets aufrufen

Klicken Sie in der Liste der Buckets in Ihrem Projekt auf den Storage-Bucket, den Sie zuvor erstellt haben. Die von Ihrem Job erstellten Ausgabedateien werden im Verzeichnis results angezeigt.

Terminal

Sehen Sie sich die Ergebnisse über Ihr Terminal oder mithilfe von Cloud Shell an.

1. Verwenden Sie den Befehl gcloud storage ls, um die Ausgabedateien aufzulisten:

   gcloud storage ls gs://BUCKET_NAME/results/outputs* --long
   

   Ersetzen Sie BUCKET_NAME durch den Namen des angegebenen Cloud Storage-Ausgabe-Buckets.

2. Verwenden Sie den Befehl gcloud storage cat, um die Ergebnisse in den Ausgabedateien aufzurufen:

   gcloud storage cat gs://BUCKET_NAME/results/outputs*
   

Bereinigen

Löschen Sie das Google Cloud-Projekt mit den Ressourcen, damit Ihrem Google Cloud-Konto die auf dieser Seite verwendeten Ressourcen nicht in Rechnung gestellt werden.

Hinweis: Wenn Sie diese Kurzanleitung in einem neuen Projekt befolgt haben, können Sie das Projekt löschen.

1. Wechseln Sie in der Cloud Console zur Seite Cloud Storage-Buckets.

   Buckets aufrufen

2. Klicken Sie auf das Kästchen neben dem Bucket, der gelöscht werden soll.
3. Klicken Sie zum Löschen des Buckets auf Löschen und folgen Sie der Anleitung.

4. Wenn Sie Ihr Projekt beibehalten, widerrufen Sie die Rollen, die Sie dem Compute Engine-Standarddienstkonto zugewiesen haben. Führen Sie den folgenden Befehl für jede der folgenden IAM-Rollen einmal aus:

   • roles/dataflow.admin
   • roles/dataflow.worker
   • roles/storage.objectAdmin

   gcloud projects remove-iam-policy-binding PROJECT_ID \
       --member=serviceAccount:[e-mail] \
       --role=SERVICE_ACCOUNT_ROLE

5. Optional: Widerrufen Sie die von Ihnen erstellten Anmeldedaten für die Authentifizierung und löschen Sie die lokale Datei mit den Anmeldedaten:

   gcloud auth application-default revoke

6. Optional: Widerrufen Sie Anmeldedaten von der gcloud-CLI.

   gcloud auth revoke

Nächste Schritte

• Programmiermodell für Apache Beam
• Pipelineoptionen festlegen
• Referenz zu Pipelineoptionen
• Pipeline bereitstellen