Wiederverwendbare Pipeline erstellen

In dieser Anleitung wird gezeigt, wie Sie eine wiederverwendbare Pipeline erstellen, die Daten aus Cloud Storage liest, Qualitätsprüfungen durchführt und in Cloud Storage schreibt.

Wiederverwendbare Pipelines haben eine reguläre Pipelinestruktur. Sie können jedoch die Konfiguration für jeden Pipelineknoten entsprechend den von einem HTTP-Server bereitgestellten Konfigurationen ändern. Beispielsweise kann eine statische Pipeline Daten aus Cloud Storage lesen, Transformationen anwenden und in eine BigQuery-Ausgabetabelle schreiben. Wenn Sie die Transformation und die BigQuery-Ausgabetabelle basierend auf der Cloud Storage-Datei, die die Pipeline liest, ändern möchten, erstellen Sie eine wiederverwendbare Pipeline.

Ziele

  • Verwenden Sie das Plug-in Argument Setter, damit die Pipeline bei jeder Ausführung andere Eingaben lesen kann.
  • Verwenden Sie das Argument Setter-Plug-in, damit die Pipeline bei jeder Ausführung verschiedene Qualitätsprüfungen vornehmen kann.
  • Schreiben Sie die Ausgabedaten jeder Ausführung in Cloud Storage.

Kosten

In dieser Anleitung werden kostenpflichtige Komponenten von Google Cloud verwendet, darunter:

  • Cloud Data Fusion
  • Cloud Storage

Sie können mithilfe des Preisrechners die Kosten für Ihre voraussichtliche Nutzung kalkulieren. Neuen Google Cloud-Nutzern steht möglicherweise eine kostenlose Testversion zur Verfügung.

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. Wählen Sie in der Google Cloud Console auf der Seite der Projektauswahl ein Google Cloud-Projekt aus oder erstellen Sie eines.

    Zur Projektauswahl

  3. Die Abrechnung für das Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für Ihr Projekt aktiviert ist.

  4. Cloud Data Fusion, Cloud Storage, and Cloud Dataproc APIs aktivieren.

    Aktivieren Sie die APIs

  5. Erstellen Sie eine Cloud Data Fusion-Instanz.

Bei Verwendung von Cloud Data Fusion verwenden Sie sowohl die Cloud Console als auch die separate UI von Cloud Data Fusion. In der Cloud Console können Sie ein Cloud Console-Projekt sowie Cloud Data Fusion-Instanzen erstellen und löschen. In der Benutzeroberfläche von Cloud Data Fusion können Sie die verschiedenen Seiten wie Pipeline Studio oder Wrangler nutzen, um Features von Cloud Data Fusion zu nutzen.

  1. Öffnen Sie in der Cloud Console die Seite Instanzen.

    Zur Seite "Instanzen"

  2. Klicken Sie in der Spalte Aktionen für die Instanz auf den Link Instanz aufrufen. Die Benutzeroberfläche von Cloud Data Fusion wird in einem neuen Browsertab geöffnet.

Das Argument Setter-Plug-in bereitstellen

  1. Klicken Sie in der Cloud Data Fusion-Web-UI rechts oben auf Hub.

  2. Klicken Sie auf das Plug-in "Argument Setter Action" und dann auf Bereitstellen.

  3. Klicken Sie im angezeigten Fenster Bereitstellen auf Fertigstellen.

  4. Klicken Sie auf Pipeline erstellen. Die Seite Pipeline Studio wird geöffnet.

Aus Cloud Storage lesen

  1. Wählen Sie im Drop-down-Menü Quelle im linken Bereich der Seite Pipeline Studio die Option Google Cloud Storage aus.
  2. Bewegen Sie den Mauszeiger auf die Cloud Storage-Quellkarte und klicken Sie auf die Schaltfläche Attribute.
  3. Geben Sie im Feld Referenzname einen Namen ein.
  4. Geben Sie im Feld Pfad ${input.path} ein. Dieses Makro steuert, was der Cloud Storage-Eingabepfad in den verschiedenen Pipelineausführungen sein wird.
  5. Entfernen Sie im rechten Feld "Ausgabeschema" das Feld offset aus dem Ausgabeschema, indem Sie in der Zeile mit dem Offset-Feld auf das Papierkorbsymbol klicken.
  6. Klicken Sie auf Validieren, um sicherzustellen, dass keine Fehler vorliegen.
  7. Klicken Sie auf die Schaltfläche X, um das Dialogfeld "Attribute" zu schließen.

Daten transformieren

  1. Wählen Sie im linken Bereich der Seite Pipeline Studio über das Drop-down-Menü Transformieren die Option Wrangler aus.
  2. Ziehen Sie im Pipeline Studio-Canvas einen Pfeil von der Cloud Storage-Karte auf die Wrangler-Karte.
  3. Bewegen Sie den Mauszeiger auf die Wrangler-Karte und klicken Sie auf die Schaltfläche Attribute.
  4. Geben Sie in das Feld Name des Eingabefelds body ein.
  5. Geben Sie im Feld Schema den Wert ${directives} ein. Dieses Makro steuert, was die Transformationslogik in den verschiedenen Pipelineausführungen sein wird.
  6. Klicken Sie auf Validieren, um sicherzustellen, dass keine Fehler vorliegen.
  7. Klicken Sie auf die Schaltfläche X, um das Dialogfeld "Attribute" zu schließen.

In Cloud Storage schreiben

  1. Wählen Sie im Drop-down-Menü Senke im linken Bereich der Seite Pipeline Studio die Option "Cloud Storage" aus.
  2. Ziehen Sie im Pipeline Studio-Canvas einen Pfeil von der Wrangler-Karte auf die Cloud Storage-Karte, die Sie gerade hinzugefügt haben.
  3. Bewegen Sie den Mauszeiger auf die Cloud Storage-Senkenkarte und klicken Sie auf die Schaltfläche Attribute.
  4. Geben Sie im Feld Referenzname einen Namen ein.
  5. Geben Sie im Feld Pfad den Pfad eines Cloud Storage-Buckets in Ihrem Projekt ein, in den die Pipeline die Ausgabedateien schreiben kann. Wenn Sie keinen Cloud Storage-Bucket haben, erstellen Sie einen.
  6. Klicken Sie auf Validieren, um sicherzustellen, dass keine Fehler vorliegen.
  7. Klicken Sie auf die Schaltfläche X, um das Menü "Attribute" zu schließen.

Die Makroargumente festlegen

  1. Wählen Sie im Drop-down-Menü Bedingungen und Aktionen im linken Bereich der Seite Pipeline Studio das Plug-in Argument Setter aus.
  2. Ziehen Sie im Pipeline Studio-Canvas einen Pfeil von der Karte "Argument Setter" auf die Karte source von Cloud Storage.
  3. Bewegen Sie den Mauszeiger auf die Karte "Argument Setter" und klicken Sie auf die Schaltfläche Attribute.
  4. Geben Sie im Feld URL Folgendes ein:

    https://storage.googleapis.com/reusable-pipeline-tutorial/args.json
    

    Die URL entspricht einem öffentlich zugänglichen Objekt in Cloud Storage, das die folgenden Inhalte enthält:

    {
      "arguments" : [
        {
          "name": "input.path",
          "value": "gs://reusable-pipeline-tutorial/user-emails.txt"
        },
        {
          "name": "directives",
          "value": "send-to-error !dq:isEmail(body)"
        }
      ]
    }
    

    Das erste der beiden Argumente ist der Wert für input.path. Der Pfad gs://reusable-pipeline-tutorial/user-emails.txt ist ein öffentlich zugängliches Objekt in Cloud Storage, das die folgenden Testdaten enthält:

    alice@example.com
    bob@example.com
    craig@invalid@example.com
    

    Das zweite Argument ist der Wert für directives. Mit dem Wert send-to-error !dq:isEmail(body) wird Wrangler so konfiguriert, dass alle Zeilen herausgefiltert werden, die keine gültige E-Mail-Adresse sind. Beispiel: craig@invalid@example.com wird herausgefiltert.

  5. Klicken Sie auf Validieren, um sicherzustellen, dass keine Fehler vorliegen.

  6. Klicken Sie auf die Schaltfläche X, um das Menü "Attribute" zu schließen.

Ihre Pipeline bereitstellen und ausführen

  1. Klicken Sie in der oberen Leiste der Seite Pipeline Studio auf Pipeline benennen. Benennen Sie die Pipeline und klicken Sie auf Speichern.
  2. Klicken Sie auf Deploy.
  3. Klicken Sie neben Ausführen auf das Drop-down-Menü , um die Laufzeitargumente zu öffnen und die Argumente input.path und directives des Makros (Laufzeit) anzuzeigen. Lassen Sie die Wertfelder leer, um Cloud Data Fusion zu informieren, dass der Argument Setter-Knoten in der Pipeline die Werte dieser Argumente während der Laufzeit festlegt.
  4. Klicken Sie auf Ausführen.

Bereinigen

Damit Ihrem Google Cloud-Konto die in dieser Anleitung verwendeten Ressourcen nicht in Rechnung gestellt werden, löschen Sie entweder das Projekt, das die Ressourcen enthält, oder Sie behalten das Projekt und löschen die einzelnen Ressourcen.

Nachdem Sie diese Anleitung abgeschlossen haben, bereinigen Sie die in Google Cloud erstellten Ressourcen, damit sie keine kostenpflichtigen Kontingente verbrauchen. In den folgenden Abschnitten erfahren Sie, wie Sie diese Ressourcen löschen oder deaktivieren.

Löschen Sie die Cloud Data Fusion-Instanz.

Folgen Sie der Anleitung zum Löschen Ihrer Cloud Data Fusion-Instanz.

Projekt löschen

Am einfachsten vermeiden Sie weitere Kosten durch Löschen des für die Anleitung erstellten Projekts.

So löschen Sie das Projekt:

  1. Wechseln Sie in der Cloud Console zur Seite Ressourcen verwalten.

    Zur Seite „Ressourcen verwalten“

  2. Wählen Sie in der Projektliste das Projekt aus, das Sie löschen möchten, und klicken Sie dann auf Löschen.
  3. Geben Sie im Dialogfeld die Projekt-ID ein und klicken Sie auf Shut down (Beenden), um das Projekt zu löschen.

Nächste Schritte