Cloud Composer 1 Cloud Composer 2
In dieser Anleitung wird gezeigt, wie Sie mit Cloud Composer einen Apache Airflow-DAG erstellen. Der DAG führt Daten aus einem öffentlichen BigQuery-Dataset mit einer CSV-Datei zusammen, die in einem Cloud Storage-Bucket gespeichert ist, und führt dann einen Dataproc Serverless-Batchjob aus, um die verknüpften Daten zu verarbeiten.
Das öffentliche BigQuery-Dataset in dieser Anleitung ist ghcn_d, eine integrierte Datenbank mit globalen Klimazusammenfassungen. Die CSV-Datei enthält Informationen zu den Daten und Namen von Feiertagen in den USA von 1997 bis 2021.
Die Frage, die wir mit dem DAG beantworten möchten, lautet: „Wie warm war es in Chicago am Thanksgiving in den letzten 25 Jahren?“
Lernziele
- Cloud Composer-Umgebung in der Standardkonfiguration erstellen
- Leeres BigQuery-Dataset erstellen
- Neuen Cloud Storage-Bucket erstellen
- Erstellen Sie einen DAG, der die folgenden Aufgaben enthält, und führen Sie ihn aus:
- Externes Dataset aus Cloud Storage in BigQuery laden
- Zwei Datasets in BigQuery verknüpfen
- PySpark-Data-Analytics-Job ausführen
Hinweise
APIs aktivieren
Aktivieren Sie folgende APIs:
Console
Enable the Dataproc, Cloud Composer, BigQuery, Cloud Storage APIs.
gcloud
Enable the Dataproc, Cloud Composer, BigQuery, Cloud Storage APIs:
gcloud services enable dataproc.googleapis.comcomposer.googleapis.com bigquery.googleapis.com storage.googleapis.com
Berechtigungen erteilen
Weisen Sie Ihrem Nutzerkonto die folgenden Rollen und Berechtigungen zu:
Gewähren Sie Rollen zum Verwalten von Cloud Composer-Umgebungen und Umgebungs-Buckets.
Weisen Sie die Rolle BigQuery-Dateninhaber (
roles/bigquery.dataOwner
) zu, um ein BigQuery-Dataset zu erstellen.Gewähren Sie die Rolle Storage Admin (
roles/storage.admin
) zum Erstellen eines Cloud Storage-Buckets.
Cloud Composer-Umgebung erstellen und vorbereiten
Erstellen Sie eine Cloud Composer-Umgebung mit Standardparametern:
- Wählen Sie eine Region in den USA aus.
- Wählen Sie die neueste Cloud Composer-Version aus.
Weisen Sie dem Dienstkonto, das in Ihrer Cloud Composer-Umgebung verwendet wird, die folgenden Rollen zu, damit die Airflow-Worker DAG-Aufgaben erfolgreich ausführen können:
- BigQuery-Nutzer (
roles/bigquery.user
) - BigQuery-Dateninhaber (
roles/bigquery.dataOwner
) - Dienstkontonutzer (
roles/iam.serviceAccountUser
) - Dataproc-Editor (
roles/dataproc.editor
) - Dataproc-Worker (
roles/dataproc.worker
)
- BigQuery-Nutzer (
Zugehörige Ressourcen erstellen
Erstellen Sie ein leeres BigQuery-Dataset mit den folgenden Parametern:
- Name:
holiday_weather
- Region:
US
- Name:
Erstellen Sie einen neuen Cloud Storage-Bucket am multiregionalen Standort
US
.Führen Sie den folgenden Befehl aus, um den privaten Google-Zugriff im Standardsubnetz der Region zu aktivieren, in der Sie Dataproc Serverless ausführen möchten, um die Netzwerkanforderungen zu erfüllen. Wir empfehlen, dieselbe Region wie in Ihrer Cloud Composer-Umgebung zu verwenden.
gcloud compute networks subnets update default \ --region DATAPROC_SERVERLESS_REGION \ --enable-private-ip-google-access
Datenverarbeitung mit Dataproc Serverless
PySpark-Beispiel-Job ansehen
Der folgende Code ist ein PySpark-Beispieljob, der die Temperatur von Zehntelgrad Celsius in Grad Celsius umwandelt. Dieser Job wandelt Temperaturdaten aus dem Dataset in ein anderes Format um.
Unterstützende Dateien in Cloud Storage hochladen
So laden Sie die PySpark-Datei und das in holidays.csv
gespeicherte Dataset hoch:
Speichern Sie data_analytics_process.py auf Ihrem lokalen Computer.
Speichern Sie holidays.csv auf Ihrem lokalen Computer.
Rufen Sie in der Google Cloud Console die Seite Cloud Storage-Browser auf:
Klicken Sie auf den Namen des Buckets, den Sie zuvor erstellt haben.
Klicken Sie im Tab Objekte für den Bucket auf die Schaltfläche Dateien hochladen, wählen Sie im angezeigten Dialogfeld
data_analytics_process.py
undholidays.csv
aus und klicken Sie auf Öffnen.
Datenanalyse-DAG
Beispiel-DAG ansehen
Der DAG verwendet mehrere Operatoren, um die Daten zu transformieren und zu vereinheitlichen:
Mit
GCSToBigQueryOperator
wird die Datei holidays.csv aus Cloud Storage in eine neue Tabelle im BigQuery-Datasetholidays_weather
aufgenommen, das Sie zuvor erstellt haben.Der
DataprocCreateBatchOperator
erstellt und führt einen PySpark-Batchjob mit Dataproc Serverless aus.Mit
BigQueryInsertJobOperator
werden die Daten aus holidays.csv in der Spalte „Date“ mit Wetterdaten aus dem öffentlichen BigQuery-Dataset ghcn_d zusammengeführt. DieBigQueryInsertJobOperator
-Aufgaben werden mit einer For-Schleife dynamisch generiert. Zur besseren Lesbarkeit in der Grafikansicht der Airflow-UI befinden sich diese Aufgaben in einerTaskGroup
.
Variablen über die Airflow-UI hinzufügen
In Airflow sind Variablen eine universelle Möglichkeit, beliebige Einstellungen oder Konfigurationen als einfachen Speicher für Schlüssel/Wert-Paare zu speichern und abzurufen. Dieser DAG verwendet Airflow-Variablen zum Speichern gängiger Werte. So fügen Sie sie zu Ihrer Umgebung hinzu:
Rufen Sie über die Cloud Composer-Konsole auf die Airflow-UI auf.
Klicken Sie auf Verwaltung > Variablen.
Fügen Sie die folgenden Variablen hinzu:
gcp_project
: Ihre Projekt-ID.gcs_bucket
: der Name des zuvor erstellten Buckets (ohne das Präfixgs://
).gce_region
: Die Region, in der Ihr Dataproc-Job ausgeführt werden soll, der die Anforderungen an das serverlose Dataproc-Netzwerk erfüllt. Das ist die Region, in der Sie zuvor den privaten Google-Zugriff aktiviert haben.dataproc_service_account
: das Dienstkonto für Ihre Cloud Composer-Umgebung. Sie finden dieses Dienstkonto auf dem Tab für die Umgebungskonfiguration für Ihre Cloud Composer-Umgebung.
DAG in den Bucket Ihrer Umgebung hochladen
Cloud Composer plant DAGs, die sich im Ordner /dags
im Bucket Ihrer Umgebung befinden. So laden Sie den DAG über die Google Cloud Console hoch:
Speichern Sie data_analytics_dag.py auf Ihrem lokalen Computer.
Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.
Klicken Sie in der Liste der Umgebungen in der Spalte DAG-Ordner auf den Link DAGs. Der DAGs-Ordner Ihrer Umgebung wird geöffnet.
Klicken Sie auf Dateien hochladen.
Wählen Sie
data_analytics_dag.py
auf Ihrem lokalen Computer aus und klicken Sie auf Öffnen.
DAG auslösen
Klicken Sie in Ihrer Cloud Composer-Umgebung auf den Tab DAGs.
Klicken Sie auf die DAG-ID
data_analytics_dag
.Klicken Sie auf Trigger DAG (DAG auslösen).
Warten Sie etwa fünf bis zehn Minuten, bis Sie ein grünes Häkchen sehen, das anzeigt, dass die Aufgaben erfolgreich abgeschlossen wurden.
Erfolg des DAG validieren
Öffnen Sie in der Google Cloud Console die Seite BigQuery.
Klicken Sie im Explorer-Bereich auf den Namen Ihres Projekts.
Klicken Sie auf
holidays_weather_joined
.Klicken Sie auf „Vorschau“, um die resultierende Tabelle anzusehen. Beachten Sie, dass die Zahlen in der Wertspalte in Zehntelgrad Celsius angegeben sind.
Klicken Sie auf
holidays_weather_normalized
.Klicken Sie auf „Vorschau“, um die resultierende Tabelle anzusehen. Beachten Sie, dass die Zahlen in der Spalte Werte in Grad Celsius angegeben sind.
Details zu Dataproc Serverless (optional)
Sie können eine erweiterte Version dieses DAG mit einem komplexeren PySpark-Datenverarbeitungsablauf ausprobieren. Siehe Dataproc-Erweiterung für das Data Analytics-Beispiel auf GitHub.
Bereinigen
Löschen Sie einzelne Ressourcen, die Sie für diese Anleitung erstellt haben:
Löschen Sie den Cloud Storage-Bucket, den Sie für diese Anleitung erstellt haben.
Löschen Sie die Cloud Composer-Umgebung, einschließlich des Buckets der Umgebung.