Data Analytics-DAG in Google Cloud mit Daten aus AWS ausführen

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Diese Anleitung ist eine Änderung von DAG für Datenanalyse ausführen Google Cloud und zeigt, wie Sie Ihre Cloud Composer-Umgebung mit Amazon Web Services verbinden, um dort gespeicherte Daten zu nutzen. Außerdem wird gezeigt, wie Sie mit Cloud Composer einen Apache Airflow DAG erstellen. Der DAG verknüpft Daten aus einem öffentlichen BigQuery-Dataset und einer CSV-Datei, die in einem Amazon Web Services (AWS) S3-Bucket gespeichert ist, und führt dann einen Google Cloud Serverless for Apache Spark-Batchjob aus, um die verknüpften Daten zu verarbeiten.

Das öffentliche BigQuery-Dataset in dieser Anleitung ist ghcn_d, eine integrierte Datenbank mit Klimazusammenfassungen aus aller Welt. 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 an Thanksgiving in den letzten 25 Jahren?“

Ziele

• Cloud Composer-Umgebung in der Standardkonfiguration erstellen
• Bucket in AWS S3 erstellen
• Leeres BigQuery-Dataset erstellen
• Neuen Cloud Storage-Bucket erstellen
• DAG mit nachfolgenden Aufgaben erstellen und ausführen:
  • Externes Dataset aus S3 in Cloud Storage laden
  • Externes Dataset aus Cloud Storage in BigQuery laden
  • Zwei Datasets in BigQuery verknüpfen
  • PySpark-Job für Datenanalyse ausführen

Hinweis

Berechtigungen in AWS verwalten

1. Erstellen Sie ein AWS-Konto.

2. Folgen Sie der Anleitung im Abschnitt Richtlinien mit dem visuellen Editor erstellen der AWS-Anleitung IAM-Richtlinien erstellen, um eine benutzerdefinierte IAM-Richtlinie für AWS S3 mit der folgenden Konfiguration zu erstellen:

   • Dienst:S3
   • ListAllMyBuckets (s3:ListAllMyBuckets) zum Aufrufen Ihres S3-Buckets
   • CreateBucket (s3:CreateBucket) zum Erstellen eines Buckets
   • PutBucketOwnershipControls (s3:PutBucketOwnershipControls) zum Erstellen eines Buckets
   • ListBucket (s3:ListBucket) zum Erteilen 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:Keine
   • Name:TutorialPolicy

   Weitere Informationen zu den einzelnen Konfigurationen finden Sie in der Liste der in Amazon S3 unterstützten Aktionen.

3. Fügen Sie Ihrer Identität die IAM-Richtlinie TutorialPolicy hinzu.

Achtung: Wenn Sie für „Bucket“ und „Objekt“ die Option „Beliebig“ auswählen, wird die Berechtigung für alle Ressourcen in S3 erteilt. Um den Umfang dieser Richtlinie einzugrenzen, können Sie die Berechtigungen auf einen bestimmten Bucket und ein bestimmtes Objekt beschränken, nachdem Sie sie später in der Anleitung erstellt haben.

APIs aktivieren

Aktivieren Sie folgende APIs:

Console

Aktivieren Sie die Dataproc, Cloud Composer, BigQuery und Cloud Storage APIs.

Rollen, die zum Aktivieren von APIs erforderlich sind

Zum Aktivieren von APIs benötigen Sie die IAM-Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin), die die Berechtigung serviceusage.services.enable enthält. Informationen zum Zuweisen von Rollen.

APIs aktivieren

gcloud

Aktivieren Sie die Dataproc, Cloud Composer, BigQuery und Cloud Storage APIs:

Rollen, die zum Aktivieren von APIs erforderlich sind

Zum Aktivieren von APIs benötigen Sie die IAM-Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin), die die Berechtigung serviceusage.services.enable enthält. Informationen zum Zuweisen von Rollen.

gcloud services enable dataproc.googleapis.com  composer.googleapis.com  bigquery.googleapis.com  storage.googleapis.com

Berechtigungen erteilen

Erteilen Sie Ihrem Nutzerkonto die folgenden Rollen und Berechtigungen:

• Erteilen Sie Rollen zum Verwalten von Cloud Composer-Umgebungen und Umgebungs-Buckets.

• Erteilen Sie die Rolle BigQuery-Dateninhaber (roles/bigquery.dataOwner), um ein BigQuery-Dataset zu erstellen.

• Erteilen Sie die Rolle Storage-Administrator (roles/storage.admin), um einen Cloud Storage-Bucket zu erstellen.

Cloud Composer-Umgebung erstellen und vorbereiten

1. Erstellen Sie eine Cloud Composer-Umgebung mit Standard parametern:

   • Wählen Sie eine Region in den USA aus.
   • Wählen Sie die neueste Cloud Composer-Version aus.
   Hinweis: Der BigQuery-Teil dieser Anleitung muss in der US Multiregion ausgeführt werden. Wir empfehlen, für Ihre Cloud Composer-Umgebung eine Region in den USA auszuwählen, um Kosten und Latenz zu reduzieren. Die Anleitung kann aber auch ausgeführt werden, wenn sich Ihre Cloud Composer-Umgebung in einer anderen Region befindet.

2. Erteilen Sie dem in Ihrer Cloud Composer-Umgebung verwendeten Dienstkonto die folgenden Rollen, damit die Airflow-Worker DAG-Aufgaben erfolgreich ausführen können:

   • BigQuery User (roles/bigquery.user)
   • BigQuery-Dateninhaber (roles/bigquery.dataOwner)
   • Dienstkontonutzer (roles/iam.serviceAccountUser)
   • Dataproc-Bearbeiter (roles/dataproc.editor)
   • Dataproc-Worker (roles/dataproc.worker)

Verwandte Ressourcen in erstellen und ändern Google Cloud

1. Installieren Sie das apache-airflow-providers-amazon PyPI-Paket in Ihrer Cloud Composer-Umgebung.

2. Erstellen Sie ein leeres BigQuery-Dataset mit den folgenden Parametern:

   • Name: holiday_weather
   • Region: US

3. Erstellen Sie einen neuen Cloud Storage-Bucket in der US Multiregion.

4. Führen Sie den folgenden Befehl aus, um den privaten Google-Zugriff im Standardsubnetz in der Region zu aktivieren, in der Sie Google Cloud Serverless for Apache Spark ausführen möchten, um die Netzwerkanforderungenzu erfüllen. Wir empfehlen, 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
   

Achtung: Das leere BigQuery-Dataset muss sich in der Region „USA“ befinden, um die Anforderungen an die Colocation für die Übertragung des öffentlichen BigQuery-Datasets ghcn_d zu erfüllen.

Verwandte Ressourcen in AWS erstellen

Erstellen Sie einen S3-Bucket mit den Standardeinstellungen in Ihrer bevorzugten Region.

Verbindung von Cloud Composer zu AWS herstellen

Hinweis: Wir empfehlen, alle Anmeldedaten für Verbindungen in Secret Manager zu speichern. Wenn Sie diese Anleitung für Produktionszwecke verwenden möchten, sollten Sie Secret Manager für Ihre Umgebung konfigurieren. Weitere Informationen finden Sie unter Secret Manager für Ihre Umgebung konfigurieren.

1. AWS-Zugriffsschlüssel-ID und geheimen Zugriffsschlüssel abrufen

2. Fügen Sie Ihre AWS S3 Verbindung über die Airflow-Benutzeroberfläche hinzu:

   1. Klicken Sie auf Admin > Verbindungen.

   2. Erstellen Sie eine neue Verbindung mit der folgenden Konfiguration:

      • Verbindungs-ID:aws_s3_connection
      • Verbindungstyp:Amazon S3
      • Extras (oder JSON für zusätzliche Felder): {"aws_access_key_id":"your_aws_access_key_id", "aws_secret_access_key": "your_aws_secret_access_key"}

Datenverarbeitung mit Google Cloud Serverless for Apache Spark

In diesem Abschnitt wird die Verarbeitung von Daten mit Google Cloud Serverless for Apache Spark beschrieben.

Beispiel für einen PySpark-Job ansehen

Der unten gezeigte Code ist ein Beispiel für einen PySpark-Job, der die Temperatur von Zehntelgrad Celsius in Grad Celsius umwandelt. Mit diesem Job werden Temperaturdaten aus dem Dataset in ein anderes Format konvertiert.

import sys


from py4j.protocol import Py4JJavaError
from pyspark.sql import SparkSession
from pyspark.sql.functions import col


if __name__ == "__main__":
    BUCKET_NAME = sys.argv[1]
    READ_TABLE = sys.argv[2]
    WRITE_TABLE = sys.argv[3]

    # Create a SparkSession, viewable via the Spark UI
    spark = SparkSession.builder.appName("data_processing").getOrCreate()

    # Load data into dataframe if READ_TABLE exists
    try:
        df = spark.read.format("bigquery").load(READ_TABLE)
    except Py4JJavaError as e:
        raise Exception(f"Error reading {READ_TABLE}") from e

    # Convert temperature from tenths of a degree in celsius to degrees celsius
    df = df.withColumn("value", col("value") / 10)
    # Display sample of rows
    df.show(n=20)

    # Write results to GCS
    if "--dry-run" in sys.argv:
        print("Data will not be uploaded to BigQuery")
    else:
        # Set GCS temp location
        temp_path = BUCKET_NAME

        # Saving the data to BigQuery using the "indirect path" method and the spark-bigquery connector
        # Uses the "overwrite" SaveMode to ensure DAG doesn't fail when being re-run
        # See https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html#save-modes
        # for other save mode options
        df.write.format("bigquery").option("temporaryGcsBucket", temp_path).mode(
            "overwrite"
        ).save(WRITE_TABLE)
        print("Data written to BigQuery")

PySpark-Datei in Cloud Storage hochladen

So laden Sie die PySpark-Datei in Cloud Storage hoch:

1. Speichern Sie data_analytics_process.py auf Ihrem lokalen Computer.

2. Rufen Sie in der Google Cloud Console die Seite Cloud Storage-Browser auf:

   Zum Cloud Storage Browser

3. Klicken Sie auf den Namen des Buckets, den Sie zuvor erstellt haben.

4. 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:

1. Speichern Sie holidays.csv auf Ihrem lokalen Computer.
2. Folgen Sie der AWS-Anleitung, um die Datei in Ihren Bucket hochzuladen.

DAG für Datenanalyse

In diesem Abschnitt wird beschrieben, wie Sie den DAG für die Datenanalyse konfigurieren und verwenden.

Beispiel-DAG ansehen

Der DAG verwendet mehrere Operatoren, um die Daten zu transformieren und zu vereinheitlichen:

• Der S3ToGCSOperator überträgt die Datei holidays.csv aus Ihrem AWS S3-Bucket in Ihren Cloud Storage-Bucket.

• Der GCSToBigQueryOperator nimmt die Datei holidays.csv aus Cloud Storage in eine neue Tabelle im BigQuery-Dataset holidays_weather auf, das Sie zuvor erstellt haben.

• Der DataprocCreateBatchOperator erstellt und führt einen PySpark-Batchjob mit Serverless for Apache Spark aus.

• Der BigQueryInsertJobOperator verknüpft die Daten aus holidays.csv in der Spalte „Date“ mit Wetterdaten aus dem öffentlichen BigQuery-Dataset ghcn_d. Die BigQueryInsertJobOperator Aufgaben werden dynamisch mit einer for-Schleife generiert und befinden sich in einer TaskGroup, um die Lesbarkeit in der Diagramm Ansicht der Airflow-Benutzeroberfläche zu verbessern.

import datetime

from airflow import models
from airflow.providers.google.cloud.operators import dataproc
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import (
    GCSToBigQueryOperator,
)
from airflow.providers.google.cloud.transfers.s3_to_gcs import S3ToGCSOperator
from airflow.utils.task_group import TaskGroup

PROJECT_NAME = "{{var.value.gcp_project}}"
REGION = "{{var.value.gce_region}}"

# BigQuery configs
BQ_DESTINATION_DATASET_NAME = "holiday_weather"
BQ_DESTINATION_TABLE_NAME = "holidays_weather_joined"
BQ_NORMALIZED_TABLE_NAME = "holidays_weather_normalized"

# Dataproc configs
BUCKET_NAME = "{{var.value.gcs_bucket}}"
PYSPARK_JAR = "gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar"
PROCESSING_PYTHON_FILE = f"gs://{BUCKET_NAME}/data_analytics_process.py"

# S3 configs
S3_BUCKET_NAME = "{{var.value.s3_bucket}}"

BATCH_ID = "data-processing-{{ ts_nodash | lower}}"  # Dataproc serverless only allows lowercase characters
BATCH_CONFIG = {
    "pyspark_batch": {
        "jar_file_uris": [PYSPARK_JAR],
        "main_python_file_uri": PROCESSING_PYTHON_FILE,
        "args": [
            BUCKET_NAME,
            f"{BQ_DESTINATION_DATASET_NAME}.{BQ_DESTINATION_TABLE_NAME}",
            f"{BQ_DESTINATION_DATASET_NAME}.{BQ_NORMALIZED_TABLE_NAME}",
        ],
    },
    "environment_config": {
        "execution_config": {
            "service_account": "{{var.value.dataproc_service_account}}"
        }
    },
}

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1), datetime.datetime.min.time()
)

default_dag_args = {
    # Setting start date as yesterday starts the DAG immediately when it is
    # detected in the Cloud Storage bucket.
    "start_date": yesterday,
    # To email on failure or retry set 'email' arg to your email and enable
    # emailing here.
    "email_on_failure": False,
    "email_on_retry": False,
}

with models.DAG(
    "s3_to_gcs_dag",
    # Continue to run DAG once per day
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:
    s3_to_gcs_op = S3ToGCSOperator(
        task_id="s3_to_gcs",
        bucket=S3_BUCKET_NAME,
        gcp_conn_id="google_cloud_default",
        aws_conn_id="aws_s3_connection",
        dest_gcs=f"gs://{BUCKET_NAME}",
    )

    create_batch = dataproc.DataprocCreateBatchOperator(
        task_id="create_batch",
        project_id=PROJECT_NAME,
        region=REGION,
        batch=BATCH_CONFIG,
        batch_id=BATCH_ID,
    )

    load_external_dataset = GCSToBigQueryOperator(
        task_id="run_bq_external_ingestion",
        bucket=BUCKET_NAME,
        source_objects=["holidays.csv"],
        destination_project_dataset_table=f"{BQ_DESTINATION_DATASET_NAME}.holidays",
        source_format="CSV",
        schema_fields=[
            {"name": "Date", "type": "DATE"},
            {"name": "Holiday", "type": "STRING"},
        ],
        skip_leading_rows=1,
        write_disposition="WRITE_TRUNCATE",
    )

    with TaskGroup("join_bq_datasets") as bq_join_group:
        for year in range(1997, 2022):
            BQ_DATASET_NAME = f"bigquery-public-data.ghcn_d.ghcnd_{str(year)}"
            BQ_DESTINATION_TABLE_NAME = "holidays_weather_joined"
            # Specifically query a Chicago weather station
            WEATHER_HOLIDAYS_JOIN_QUERY = f"""
            SELECT Holidays.Date, Holiday, id, element, value
            FROM `{PROJECT_NAME}.holiday_weather.holidays` AS Holidays
            JOIN (SELECT id, date, element, value FROM {BQ_DATASET_NAME} AS Table
            WHERE Table.element="TMAX" AND Table.id="USW00094846") AS Weather
            ON Holidays.Date = Weather.Date;
            """

            # For demo purposes we are using WRITE_APPEND
            # but if you run the DAG repeatedly it will continue to append
            # Your use case may be different, see the Job docs
            # https://cloud.google.com/bigquery/docs/reference/rest/v2/Job
            # for alternative values for the writeDisposition
            # or consider using partitioned tables
            # https://cloud.google.com/bigquery/docs/partitioned-tables
            bq_join_holidays_weather_data = BigQueryInsertJobOperator(
                task_id=f"bq_join_holidays_weather_data_{str(year)}",
                configuration={
                    "query": {
                        "query": WEATHER_HOLIDAYS_JOIN_QUERY,
                        "useLegacySql": False,
                        "destinationTable": {
                            "projectId": PROJECT_NAME,
                            "datasetId": BQ_DESTINATION_DATASET_NAME,
                            "tableId": BQ_DESTINATION_TABLE_NAME,
                        },
                        "writeDisposition": "WRITE_APPEND",
                    }
                },
                location="US",
            )

        s3_to_gcs_op >> load_external_dataset >> bq_join_group >> create_batch

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. Dieser DAG verwendet Airflow-Variablen, um allgemeine Werte zu speichern. So fügen Sie sie Ihrer Umgebung hinzu:

1. Rufen Sie die Airflow-Benutzeroberfläche über Google Cloud die Console auf.

2. Klicken Sie auf Admin > Variablen.

3. Fügen Sie die folgenden Variablen hinzu:

   • s3_bucket: der Name des zuvor erstellten S3-Buckets.

   • gcp_project: Ihre Projekt-ID.

   • gcs_bucket: der Name des zuvor erstellten Buckets (ohne das Präfix gs://).

   • gce_region: die Region, in der Sie Ihren Dataproc-Job ausführen möchten, der die Google Cloud Netzwerkanforderungen von Serverless for Apache Spark erfüllt. Dies ist die Region, in der Sie zuvor den privater Google-Zugriff 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 Ordner /dags im Bucket Ihrer Umgebung befinden. So laden Sie den DAG über die Google Cloud Console hoch:

1. Speichern Sie s3togcsoperator_tutorial.py auf Ihrem lokalen Computer.

2. Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.

   Zur Seite Umgebungen

3. Klicken Sie in der Liste der Umgebungen in der Spalte DAG-Ordner auf den Link DAGs. Der DAGs-Ordner Ihrer Umgebung wird geöffnet.

4. Klicken Sie auf Dateien hochladen.

5. Wählen Sie s3togcsoperator_tutorial.py auf Ihrem lokalen Computer aus und klicken Sie auf Öffnen.

DAG auslösen

1. Klicken Sie in Ihrer Cloud Composer-Umgebung auf den Tab DAGs.

2. Klicken Sie auf die DAG-ID s3_to_gcs_dag.

3. Klicken Sie auf DAG auslösen.

4. Warten Sie etwa fünf bis zehn Minuten, bis ein grünes Häkchen angezeigt wird, das darauf hinweist, dass die Aufgaben erfolgreich abgeschlossen wurden.

Erfolg des DAG prüfen

1. Rufen Sie in der Google Cloud Console die Seite BigQuery auf.

   BigQuery aufrufen

2. Klicken Sie im Bereich Explorer auf den Namen Ihres Projekts.

3. Klicken Sie auf holidays_weather_joined.

4. Klicken Sie auf „Vorschau“, um die resultierende Tabelle anzusehen. Beachten Sie, dass die Zahlen in der Spalte „Wert“ in Zehntelgrad Celsius angegeben sind.

5. Klicken Sie auf holidays_weather_normalized.

6. Klicken Sie auf „Vorschau“, um die resultierende Tabelle anzusehen. Beachten Sie, dass die Zahlen in der Spalte „Wert“ in Grad Celsius angegeben sind.

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 von Ihnen erstellten AWS S3-Bucket.

• Löschen Sie den Cloud Storage-Bucket, den Sie für diese Anleitung erstellt haben.

• Löschen Sie das BigQuery-Dataset.

• Löschen Sie die Cloud Composer-Umgebung, einschließlich des Buckets der Umgebung, den Sie manuell löschen müssen.

Nächste Schritte

• DAG für Datenanalyse in ausführen Google Cloud.
• DAG für Datenanalyse in Azure ausführen.