Datenanalyse-DAG in Google Cloud mit Daten aus Azure ausführen

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Dieses Tutorial ist eine Modifikation von Führen Sie einen Datenanalyse-DAG in Google Cloud aus, der zeigt, wie Sie Ihre Cloud Composer-Umgebung mit Microsoft Azure verbinden, um dort gespeicherte Daten zu nutzen. Es zeigt, wie Sie zum Erstellen eines Apache Airflow-DAG Die DAG verknüpft Daten aus einem öffentlichen BigQuery-Dataset mit einer gespeicherten CSV-Datei in einem Azure Blob Storage und führt dann einen Dataproc Serverless-Batchjob aus, um die verknüpften Daten.

Das öffentliche BigQuery-Dataset in dieser Anleitung ist ghcn_d, einer integrierten Datenbank mit Klimazusammenfassungen auf der ganzen Welt. Die CSV-Datei enthält Informationen zu den Daten und Namen von US-Feiertagen von 1997 bis 2021.

Mit dem DAG möchten wir die Frage beantworten: „Wie warm war es in Chicago an Thanksgiving in den letzten 25 Jahren?“

Hinweis: In dieser Anleitung wird Airflow 2 verwendet.

Lernziele

• Cloud Composer-Umgebung in der Standardkonfiguration erstellen
• Blob in Azure erstellen
• Leeres BigQuery-Dataset erstellen
• Neuen Cloud Storage-Bucket erstellen
• Erstellen und führen Sie einen DAG mit den folgenden Aufgaben aus:
  • Externes Dataset aus Azure Blob Storage laden in Cloud Storage
  • Externes Dataset aus Cloud Storage in BigQuery laden
  • Zwei Datasets in BigQuery verknüpfen
  • PySpark-Job für Datenanalyse ausführen

Hinweise

APIs aktivieren

Aktivieren Sie folgende APIs:

Console

Enable the Dataproc, Cloud Composer, BigQuery, Cloud Storage APIs.

Enable the APIs

gcloud

Enable the Dataproc, Cloud Composer, BigQuery, Cloud Storage APIs:

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

Berechtigungen erteilen

Gewähren Sie Ihrem Nutzerkonto die folgenden Rollen und Berechtigungen:

• Gewähren Sie Rollen zum Verwalten von Cloud Composer-Umgebungen und Umgebungs-Buckets.

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

• Gewähren 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 den Standardparametern:

   • 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, eine Region in den USA für Ihre Cloud Composer-Umgebung auszuwählen, um Kosten und Latenz zu reduzieren. Das Tutorial kann aber auch ausgeführt werden, wenn sich Ihre Cloud Composer-Umgebung in einer anderen Region befindet.

2. 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)

Zugehörige Ressourcen in Google Cloud erstellen und ändern

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

2. Leeres BigQuery-Dataset erstellen mit den folgenden Parametern:

   • Name: holiday_weather
   • Region: US

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

4. 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. Mi. empfehlen, dieselbe Region wie in Cloud Composer zu verwenden zu verbessern.

   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 Co-Lokalisierung für die Übertragung des öffentlichen BigQuery-Datasets ghcn_d zu erfüllen.

Zugehörige Ressourcen in Azure erstellen

1. Speicherplatzkonto erstellen mit den Standardeinstellungen.

2. Zugriffsschlüssel und Verbindungsstring abrufen für Ihr Speicherkonto.

3. Container erstellen mit den Standardoptionen in Ihrem neu erstellten Speicherkonto.

4. Weisen Sie dem Container, den Sie im vorherigen Schritt erstellt haben, die Rolle Storage Blob Delegator zu.

5. Laden Sie holidays.csv hoch, um mit den Standardoptionen im Azure-Portal einen Block-Blob zu erstellen.

6. SAS-Token erstellen für das Block-Blob, das Sie im vorherigen Schritt im Azure-Portal erstellt haben.

   • Signaturmethode: Schlüssel für die Nutzerdelegierung
   • Berechtigungen: Lesen
   • Zulässige IP-Adresse: Keine
   • Zulässige Protokolle: nur HTTPS

Hinweis: SAS-Tokens sind eine sichere und dokumentierte Möglichkeit, auf Ihr Azure-Speicherkonto zuzugreifen. Sie sind jedoch zeitkritisch und haben ein Ablaufdatum. Wählen Sie daher ein entsprechendes Ablaufdatum aus.

Über Cloud Composer eine Verbindung zu Azure herstellen

Hinweis: Es empfiehlt sich, alle Anmeldedaten für Verbindungen in Secret Manager Wenn Sie diese Anleitung für Produktionszwecke verwenden möchten, sollten Sie Secret Manager für Ihre Umgebung konfigurieren. Weitere Informationen finden Sie unter Konfigurieren Sie Secret Manager für Ihre Umgebung.

So fügen Sie über die Airflow-Benutzeroberfläche eine Microsoft Azure-Verbindung hinzu:

1. Klicken Sie auf Verwaltung > Verbindungen.

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

   • Verbindungs-ID: azure_blob_connection
   • Verbindungstyp: Azure Blob Storage
   • Blob Storage Login (Blob-Speicheranmeldung): Name Ihres Speicherkontos
   • Blob Storage-Schlüssel: Der Zugriffsschlüssel für Ihr Speicherkonto.
   • Blob Storage Account Connection String:Ihr Speicherkonto Verbindungsstring
   • SAS-Token: Das SAS-Token, das aus Ihrem Blob generiert wurde

Datenverarbeitung mit Dataproc Serverless

PySpark-Beispieljob ansehen

Der folgende Code ist ein Beispiel für einen PySpark-Job, mit dem Temperaturen von Zehntelgraden Celsius in Grad Celsius umgewandelt werden. Mit diesem Job werden Temperaturdaten aus dem Datensatz 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 des Buckets auf die Schaltfläche Dateien hochladen. wählen Sie data_analytics_process.py im angezeigten Dialogfeld 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 AzureBlobStorageToGCSOperator wird die Datei holidays.csv aus Ihrem Azure-Blockblob in Ihren Cloud Storage-Bucket übertragen.

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

• Mit DataprocCreateBatchOperator wird ein PySpark-Batchjob mit Dataproc Serverless erstellt und ausgeführt.

• Die BigQueryInsertJobOperator werden die Daten aus holidays.csv „Datum“ Spalte mit Wetterdaten aus dem öffentlichen BigQuery-Dataset ghcn_d verwendet. Die BigQueryInsertJobOperator-Aufgaben sind dynamisch mit einer For-Schleife generiert, und diese Aufgaben befinden sich TaskGroup für eine bessere Lesbarkeit in der Grafikansicht der Airflow-UI.

Wichtig: Die DataprocCreateBatchOperator ist in den Paketversionen 6.2.0 und höher von apache-airflow-providers-google verfügbar. Die erforderlichen Versionen dieses Pakets sind ab Cloud Composer-Version 1.17.10 und 2.0.3 in Ihrer Umgebung vorinstalliert.

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.microsoft.azure.transfers.azure_blob_to_gcs import (
    AzureBlobStorageToGCSOperator,
)
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"

# Azure configs
AZURE_BLOB_NAME = "{{var.value.azure_blob_name}}"
AZURE_CONTAINER_NAME = "{{var.value.azure_container_name}}"

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(
    "azure_to_gcs_dag",
    # Continue to run DAG once per day
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:
    azure_blob_to_gcs = AzureBlobStorageToGCSOperator(
        task_id="azure_blob_to_gcs",
        # Azure args
        blob_name=AZURE_BLOB_NAME,
        container_name=AZURE_CONTAINER_NAME,
        wasb_conn_id="azure_blob_connection",
        filename=f"https://console.cloud.google.com/storage/browser/{BUCKET_NAME}/",
        # GCP args
        gcp_conn_id="google_cloud_default",
        object_name="holidays.csv",
        bucket_name=BUCKET_NAME,
        gzip=False,
        impersonation_chain=None,
    )

    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",
            )

        azure_blob_to_gcs >> 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 einfaches Schlüssel/Wert-Verzeichnis zu speichern und abzurufen. In diesem DAG werden Airflow-Variablen verwendet, um gängige Werte zu speichern. So fügen Sie sie Ihrer Umgebung hinzu:

1. Über die Cloud Composer-Konsole auf die Airflow-Benutzeroberfläche zugreifen

2. Klicken Sie auf Admin > Variablen.

3. Fügen Sie die folgenden Variablen hinzu:

   • gcp_project: Ihre Projekt-ID.

   • gcs_bucket: der Name des Buckets, den Sie zuvor erstellt haben (ohne das Präfix gs://).

   • gce_region: die Region, in der Sie Ihren Dataproc-Job ausführen möchten, der die Anforderungen an die Dataproc Serverless-Netzwerkkonfiguration erfüllt. Das ist die Region, in der Sie den privaten Google-Zugriff zuvor aktiviert haben.

   • dataproc_service_account: das Dienstkonto für Ihr Cloud Composer-Umgebung. Sie finden diesen Dienst auf dem Tab für die Umgebungskonfiguration Cloud Composer-Umgebung.

   • azure_blob_name: der Name des Blobs, das Sie zuvor erstellt haben.

   • azure_container_name: der Name des Containers, den Sie zuvor erstellt haben.

DAG in den Bucket Ihrer Umgebung hochladen

Cloud Composer plant DAGs, die sich in der /dags-Ordner im Bucket Ihrer Umgebung. So laden Sie den DAG mit der Google Cloud Console hoch:

1. Speichern Sie azureblobstoretogcsoperator_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 azureblobstoretogcsoperator_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 azure_blob_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 signalisiert, dass die Aufgaben erfolgreich abgeschlossen wurden.

Erfolg des DAG validieren

1. Öffnen Sie in der Google Cloud Console die Seite BigQuery.

   BigQuery aufrufen

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

3. Klicken Sie auf holidays_weather_joined.

4. Klicken Sie auf „Vorschau“, um die resultierende Tabelle zu sehen. Beachten Sie, dass die Zahlen in den Wertspalte in Zehntelgrad Celsius angegeben.

5. Klicken Sie auf holidays_weather_normalized.

6. Klicken Sie auf „Vorschau“, um die resultierende Tabelle zu sehen. Beachten Sie, dass die Zahlen in den Wertspalte sind in Grad Celsius angegeben.

Bereinigen

Löschen Sie die einzelnen Ressourcen, die Sie für diese Anleitung erstellt haben:

• Löschen Sie den in Azure erstellten Container.

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

• Löschen Sie das BigQuery-Dataset.

• Löschen Sie die Cloud Composer-Umgebung, einschließlich den Bucket der Umgebung manuell löschen.

Nächste Schritte

• Data Analytics-DAG in Google Cloud ausführen
• Datenanalyse-DAG in AWS ausführen