Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1
Diese Anleitung ist eine Modifikation von DAG für Datenanalyse in Google Cloud ausführen. Darin wird gezeigt, wie Sie Ihre Cloud Composer-Umgebung mit Amazon Web Services verbinden, um dort gespeicherte Daten zu nutzen. Darin wird gezeigt, wie Sie mit Cloud Composer einen Apache Airflow-DAG erstellen. Im DAG werden Daten aus einem öffentlichen BigQuery-Dataset und einer in einem Amazon Web Services (AWS) S3-Bucket gespeicherten CSV-Datei zusammengeführt. Anschließend wird ein Dataproc Serverless-Batchjob ausgeführt, um die zusammengeführten Daten zu verarbeiten.
Das öffentliche BigQuery-Dataset in dieser Anleitung ist ghcn_d, eine integrierte Datenbank mit Klimazusammenfassungen weltweit. 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 in den letzten 25 Jahren am Thanksgiving-Tag?“
Ziele
- Cloud Composer-Umgebung in der Standardkonfiguration erstellen
- Bucket in AWS S3 erstellen
- Leeres BigQuery-Dataset erstellen
- Neuen Cloud Storage-Bucket erstellen
- Erstellen und führen Sie eine DAG mit den folgenden Aufgaben aus:
- Externes Dataset von S3 in Cloud Storage laden
- Externes Dataset aus Cloud Storage in BigQuery laden
- Zwei Datasets in BigQuery zusammenführen
- PySpark-Job für die Datenanalyse ausführen
Hinweise
Berechtigungen in AWS verwalten
Folgen Sie dem Abschnitt „Richtlinien mit dem visuellen Editor erstellen“ im AWS-Tutorial zum Erstellen von IAM-Richtlinien, um eine benutzerdefinierte IAM-Richtlinie für AWS S3 mit der folgenden Konfiguration zu erstellen:
- Dienst:S3
- ListAllMyBuckets (
s3:ListAllMyBuckets
) zum Anzeigen Ihres S3-Buckets - CreateBucket (
s3:CreateBucket
) zum Erstellen eines Buckets - PutBucketOwnershipControls (
s3:PutBucketOwnershipControls
) zum Erstellen eines Buckets - ListBucket (
s3:ListBucket
) zum Gewähren der Berechtigung zum Auflisten von Objekten in einem S3-Bucket - PutObject (
s3:PutObject
) zum Hochladen von Dateien in einen Bucket - GetBucketVersioning (
s3:GetBucketVersioning
) zum Löschen eines Objekts in einem Bucket - DeleteObject (
s3:DeleteObject
) zum Löschen eines Objekts in einem Bucket - ListBucketVersions (
s3:ListBucketVersions
) zum Löschen eines Buckets - DeleteBucket (
s3:DeleteBucket
) zum Löschen eines Buckets - Ressourcen:Wählen Sie neben „Bucket“ und „Objekt“ die Option „Beliebig“ aus, um Berechtigungen für alle Ressourcen dieses Typs zu erteilen.
- Tag:Keiner
- Name:TutorialPolicy
Weitere Informationen zu den einzelnen Konfigurationen finden Sie in der Liste der in Amazon S3 unterstützten Aktionen.
Fügen Sie Ihrer Identität die IAM-Richtlinie TutorialPolicy hinzu.
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:
Rollen zum Verwalten von Cloud Composer-Umgebungen und Umgebungs-Buckets zuweisen
Weisen Sie die Rolle BigQuery-Dateninhaber (
roles/bigquery.dataOwner
) zu, um ein BigQuery-Dataset zu erstellen.Weisen Sie die Rolle Storage-Administrator (
roles/storage.admin
) zu, um einen Cloud Storage-Bucket zu erstellen.
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 aktuelle 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-Bearbeiter (
roles/dataproc.editor
) - Dataproc-Worker (
roles/dataproc.worker
)
- BigQuery-Nutzer (
Zugehörige Ressourcen in Google Clouderstellen und ändern
Installieren Sie das
apache-airflow-providers-amazon
PyPI-Paket in Ihrer Cloud Composer-Umgebung.Leeres BigQuery-Dataset erstellen mit den folgenden Parametern:
- Name:
holiday_weather
- Region:
US
- Name:
Erstellen Sie einen neuen Cloud Storage-Bucket in der Multiregion
US
.Führen Sie den folgenden Befehl aus, um privaten Google-Zugriff im Standardsubnetz in der Region zu aktivieren, in der Sie Dataproc Serverless ausführen möchten, um die Netzwerkanforderungen zu erfüllen. Es wird empfohlen, dieselbe Region wie für Ihre Cloud Composer-Umgebung zu verwenden.
gcloud compute networks subnets update default \ --region DATAPROC_SERVERLESS_REGION \ --enable-private-ip-google-access
Zugehörige Ressourcen in AWS erstellen
Erstellen Sie einen S3-Bucket mit Standardeinstellungen in der gewünschten Region.
Verbindung zu AWS über Cloud Composer herstellen
- AWS-Zugriffsschlüssel-ID und Secret-Zugriffsschlüssel abrufen
Fügen Sie Ihre AWS S3-Verbindung über die Airflow-Benutzeroberfläche hinzu:
- Klicken Sie auf Verwaltung > Verbindungen.
Erstellen Sie eine neue Verbindung mit der folgenden Konfiguration:
- Verbindungs-ID:
aws_s3_connection
- Verbindungstyp:
Amazon S3
- Extras:
{"aws_access_key_id":"your_aws_access_key_id", "aws_secret_access_key": "your_aws_secret_access_key"}
- Verbindungs-ID:
Datenverarbeitung mit Dataproc Serverless
PySpark-Beispieljob ansehen
Der unten gezeigte Code ist ein Beispiel für einen PySpark-Job, der die Temperatur von Zehntelgrad Celsius in Grad Celsius umrechnet. Mit diesem Job werden Temperaturdaten aus dem Dataset in ein anderes Format konvertiert.
PySpark-Datei in Cloud Storage hochladen
So laden Sie die PySpark-Datei in Cloud Storage hoch:
Speichern Sie data_analytics_process.py auf Ihrem lokalen Computer.
Wechseln Sie in der Google Cloud Console zur Seite Cloud Storage-Browser:
Klicken Sie auf den Namen des Buckets, den Sie zuvor erstellt haben.
Klicken Sie auf dem Tab Objekte für den Bucket auf die Schaltfläche Dateien hochladen, wählen Sie im angezeigten Dialogfeld
data_analytics_process.py
aus und klicken Sie auf Öffnen.
CSV-Datei in AWS S3 hochladen
So laden Sie die Datei holidays.csv
hoch:
- Speichern Sie
holidays.csv
auf Ihrem lokalen Computer. - Folgen Sie der AWS-Anleitung, um die Datei in Ihren Bucket hochzuladen.
DAG für die Datenanalyse
Beispiel-DAG ansehen
Im DAG werden mehrere Operatoren verwendet, um die Daten zu transformieren und zu vereinheitlichen:
Mit dem
S3ToGCSOperator
wird die Datei holidays.csv aus Ihrem AWS S3-Bucket in Ihren Cloud Storage-Bucket übertragen.Mit
GCSToBigQueryOperator
wird die Datei holidays.csv aus Cloud Storage in eine neue Tabelle im BigQuery-Datasetholidays_weather
aufgenommen, das Sie zuvor erstellt haben.Mit dem
DataprocCreateBatchOperator
wird ein PySpark-Batchjob mit Dataproc Serverless erstellt und ausgeführt.Mit
BigQueryInsertJobOperator
werden die Daten aus holidays.csv in der Spalte „Date“ mit Wetterdaten aus dem öffentlichen BigQuery-Dataset ghcn_d verknüpft. DieBigQueryInsertJobOperator
-Aufgaben werden dynamisch mit einer for-Schleife generiert und befinden sich in einemTaskGroup
, um die Lesbarkeit in der Diagrammansicht der Airflow-Benutzeroberfläche zu verbessern.
Variablen über die Airflow-Benutzeroberfläche hinzufügen
In Airflow sind Variablen eine universelle Möglichkeit, beliebige Einstellungen oder Konfigurationen als einfachen Schlüssel/Wert-Speicher zu speichern und abzurufen. In diesem DAG werden Airflow-Variablen zum Speichern allgemeiner Werte verwendet. So fügen Sie sie Ihrer Umgebung hinzu:
Über die Cloud Composer-Konsole auf die Airflow-UI zugreifen
Klicken Sie auf Verwaltung > Variablen.
Fügen Sie die folgenden Variablen hinzu:
s3_bucket
: Der Name des S3-Buckets, den Sie zuvor erstellt haben.gcp_project
: Ihre Projekt-ID.gcs_bucket
: der Name des Buckets, den Sie zuvor erstellt haben (ohne das Präfixgs://
).gce_region
: die Region, in der Sie Ihren Dataproc-Job ausführen möchten, der den Dataproc Serverless-Netzwerkanforderungen entspricht. Das ist die Region, in der Sie den privaten Google-Zugriff zuvor aktiviert haben.dataproc_service_account
: das Dienstkonto für Ihre Cloud Composer-Umgebung. Sie finden dieses Dienstkonto auf dem Tab „Umgebungskonfiguration“ für Ihre Cloud Composer-Umgebung.
DAG in den Bucket Ihrer Umgebung hochladen
Cloud Composer plant DAGs, die sich im /dags
-Ordner des Buckets Ihrer Umgebung befinden. So laden Sie den DAG über dieGoogle Cloud Console hoch:
Speichern Sie s3togcsoperator_tutorial.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
s3togcsoperator_tutorial.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
s3_to_gcs_dag
.Klicken Sie auf DAG auslösen.
Warten Sie etwa fünf bis zehn Minuten, bis ein grünes Häkchen angezeigt wird, das angibt, dass die Aufgaben erfolgreich abgeschlossen wurden.
Erfolg des DAG prüfen
Rufen Sie in der Google Cloud Console die Seite BigQuery auf.
Klicken Sie im Explorer-Bereich auf den Namen Ihres Projekts.
Klicken Sie auf
holidays_weather_joined
.Klicken Sie auf „Vorschau“, um die resultierende Tabelle aufzurufen. Die Zahlen in der Spalte „Wert“ sind in Zehntelgrad Celsius angegeben.
Klicken Sie auf
holidays_weather_normalized
.Klicken Sie auf „Vorschau“, um die resultierende Tabelle aufzurufen. Die Zahlen in der Spalte „Wert“ sind in Grad Celsius angegeben.
Bereinigen
Löschen Sie die einzelnen Ressourcen, die Sie für diese Anleitung erstellt haben:
Löschen Sie die Datei
holidays.csv
in Ihrem AWS S3-Bucket.Löschen Sie den Cloud Storage-Bucket, den Sie für diese Anleitung erstellt haben.
Löschen Sie die Cloud Composer-Umgebung und löschen Sie den Bucket der Umgebung manuell.