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.
Lernziele
- Verwenden Sie das Cloud Storage Argument Setter-Plug-in, damit die Pipeline bei jeder Ausführung andere Eingaben lesen kann.
- Verwenden Sie das Cloud Storage 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 diesem Dokument verwenden Sie die folgenden kostenpflichtigen Komponenten von Google Cloud:
- Cloud Data Fusion
- Cloud Storage
Mit dem Preisrechner können Sie eine Kostenschätzung für Ihre voraussichtliche Nutzung vornehmen.
Hinweis
- Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
Make sure that billing is enabled for your Google Cloud project.
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
Make sure that billing is enabled for your Google Cloud project.
-
Enable the Cloud Data Fusion, Cloud Storage, BigQuery, and Dataproc APIs.
- Erstellen Sie eine Cloud Data Fusion-Instanz.
Cloud Data Fusion-Weboberfläche aufrufen
Bei Verwendung von Cloud Data Fusion verwenden Sie sowohl die Google Cloud Console als auch die separate Weboberfläche von Cloud Data Fusion. In der Google Cloud Console können Sie ein Google Cloud Console-Projekt erstellen sowie Cloud Data Fusion-Instanzen erstellen und löschen. In der Weboberfläche von Cloud Data Fusion können Sie die verschiedenen Seiten wie Pipeline Studio oder Wrangler nutzen, um die Funktionen von Cloud Data Fusion zu nutzen.
Öffnen Sie in der Google Cloud Console die Seite Instanzen.
Klicken Sie in der Spalte Aktionen für die Instanz auf den Link Instanz aufrufen. Die Cloud Data Fusion-Weboberfläche wird in einem neuen Browsertab geöffnet.
Cloud Storage Argument Setter-Plug-in bereitstellen
Rufen Sie in der Cloud Data Fusion-Weboberfläche die Seite Studio auf.
Klicken Sie im Menü Aktionen auf GCS-Argument-Setter.
Aus Cloud Storage lesen
- Rufen Sie in der Cloud Data Fusion-Weboberfläche die Seite Studio auf.
- Klicken Sie auf arrow_drop_down Quelle und wählen Sie „Cloud Storage“ aus. Der Knoten für eine Cloud Storage-Quelle wird in der Pipeline angezeigt.
Klicken Sie auf dem Cloud Storage-Knoten auf Eigenschaften.
Geben Sie im Feld Referenzname einen Namen ein.
Geben Sie im Feld Pfad
${input.path}
ein. Dieses Makro steuert, was der Cloud Storage-Eingabepfad in den verschiedenen Pipelineausführungen sein wird.Entfernen Sie im rechten Bereich „Ausgabeschema“ das Feld offset aus dem Ausgabeschema, indem Sie in der Zeile mit dem Offset-Feld auf das Papierkorbsymbol klicken.
Klicken Sie auf Validieren und beheben Sie alle Fehler.
Klicken Sie auf
, um das Dialogfeld Properties (Eigenschaften) zu schließen.
Daten transformieren
- Rufen Sie in der Cloud Data Fusion-Weboberfläche auf der Seite Studio Ihre Datenpipeline auf.
- Wählen Sie im Drop-down-Menü Transformieren arrow_drop_down die Option Wrangler aus.
- Ziehen Sie im Pipeline Studio-Canvas einen Pfeil vom Cloud Storage-Knoten zum Wrangler-Knoten.
- Klicken Sie in Ihrer Pipeline auf den Wrangler-Knoten und dann auf Attribute.
- Geben Sie in das Feld Name des Eingabefelds
body
ein. - Geben Sie im Feld Schema den Wert
${directives}
ein. Dieses Makro steuert, was die Transformationslogik in den verschiedenen Pipelineausführungen sein wird. - Klicken Sie auf Validieren und beheben Sie alle Fehler.
- Klicken Sie auf , um das Dialogfeld Properties (Eigenschaften) zu schließen.
In Cloud Storage schreiben
- Rufen Sie in der Cloud Data Fusion-Weboberfläche auf der Seite Studio Ihre Datenpipeline auf.
- Wählen Sie im Drop-down-Menü Sink arrow_drop_down die Option „Cloud Storage“ aus.
- Ziehen Sie im Pipeline Studio-Canvas einen Pfeil vom Wrangler-Knoten zum Cloud Storage-Knoten, den Sie gerade hinzugefügt haben.
- Klicken Sie in Ihrer Pipeline auf den Cloud Storage-Senkenknoten und dann auf Attribute.
- Geben Sie im Feld Referenzname einen Namen ein.
- 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.
- Klicken Sie auf Validieren und beheben Sie alle Fehler.
- Klicken Sie auf , um das Dialogfeld Properties (Eigenschaften) zu schließen.
Die Makroargumente festlegen
- Rufen Sie in der Cloud Data Fusion-Weboberfläche auf der Seite Studio Ihre Datenpipeline auf.
- Klicken Sie im Drop-down-Menü arrow_drop_down Bedingungen und Aktionen auf GCS-Argument-Setter.
- Ziehen Sie im Pipeline Studio-Canvas einen Pfeil vom Knoten „Cloud Storage Argument Setter“ zum Knoten source von Cloud Storage.
- Klicken Sie in Ihrer Pipeline auf den Knoten „Cloud Storage Argument Setter“ und dann auf Attribute.
Geben Sie im Feld URL Folgendes ein:
gs://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 Pfadgs://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 Wertsend-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.Klicken Sie auf Validieren, um sicherzustellen, dass keine Fehler vorliegen.
Klicken Sie auf
, um das Dialogfeld Properties (Eigenschaften) zu schließen.
Ihre Pipeline bereitstellen und ausführen
Klicken Sie in der oberen Leiste der Seite Pipeline Studio auf Pipeline benennen. Benennen Sie die Pipeline und klicken Sie auf Speichern.
Klicken Sie auf Bereitstellen.
Klicken Sie auf das Drop-down-Menü arrow_drop_down neben Ausführen, um die Laufzeitargumente zu öffnen und die Argumente
input.path
unddirectives
des Makros (Laufzeit) aufzurufen.Lassen Sie die Wertfelder leer, um Cloud Data Fusion zu informieren, dass der Cloud Storage Argument Setter-Knoten in der Pipeline die Werte dieser Argumente während der Laufzeit festlegt.
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.
Bereinigen Sie nach Abschluss der Anleitung die inGoogle 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:
- In the Google Cloud console, go to the Manage resources page.
- In the project list, select the project that you want to delete, and then click Delete.
- In the dialog, type the project ID, and then click Shut down to delete the project.
Nächste Schritte
- Anleitungen lesen
- Weitere Anleitung durcharbeiten