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

Managed Airflow (Gen 3) | Managed Airflow (Gen 2) | Managed Airflow (Legacy Gen 1)

Diese Anleitung ist eine Änderung von DAG für Datenanalyse ausführen Google Cloud und zeigt, wie Sie Ihre Managed Airflow-Umgebung mit Microsoft Azure verbinden, um dort gespeicherte Daten zu nutzen. Außerdem wird gezeigt, wie Sie mit Managed Airflow einen Apache Airflow-DAG erstellen. Der DAG führt Daten aus einem öffentlichen BigQuery-Dataset und einer CSV-Datei zusammen, die in einem Azure Blob Storage gespeichert ist. Anschließend wird ein Batchjob von Managed Service for Apache Spark ausgeführt, um die zusammengeführten 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

  • Managed Airflow-Umgebung in der Standardkonfiguration erstellen
  • Blob in Azure erstellen
  • Leeres BigQuery-Dataset erstellen
  • Neuen Cloud Storage-Bucket erstellen
  • DAG mit den folgenden Aufgaben erstellen und ausführen:
    • Externes Dataset aus Azure Blob Storage 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

Hinweis

APIs aktivieren

Aktivieren Sie folgende APIs:

Console

Aktivieren Sie die APIs für Managed Service for Apache Spark, Managed Airflow, BigQuery und Cloud Storage.

Erforderliche Rollen zum Aktivieren von APIs

Zum Aktivieren von APIs benötigen Sie die Berechtigung serviceusage.services.enable. Wenn Sie das Projekt erstellt haben, haben Sie diese Berechtigung wahrscheinlich bereits über die Rolle „Inhaber“ (roles/owner). Andernfalls können Sie diese Berechtigung über die Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin) erhalten. Informationen zum Zuweisen von Rollen.

APIs aktivieren

gcloud

Aktivieren Sie die APIs für Managed Service for Apache Spark, Managed Airflow, BigQuery und Cloud Storage:

Erforderliche Rollen zum Aktivieren von APIs

Zum Aktivieren von APIs benötigen Sie die Berechtigung serviceusage.services.enable. Wenn Sie das Projekt erstellt haben, haben Sie diese Berechtigung wahrscheinlich bereits über die Rolle „Inhaber“ (roles/owner). Andernfalls können Sie diese Berechtigung über die Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin) erhalten. 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:

Managed Airflow-Umgebung erstellen und vorbereiten

  1. Erstellen Sie eine Managed Airflow-Umgebung mit den Standard parametern:

  2. Erteilen Sie dem Dienstkonto, das in Ihrer Managed Airflow-Umgebung verwendet wird, die folgenden Rollen, 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)

  1. Installieren Sie das apache-airflow-providers-microsoft-azure PyPI-Paket in Ihrer Managed Airflow-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 Managed Service for Apache Spark ausführen möchten, um die Netzwerkanforderungenzu erfüllen. Wir empfehlen, dieselbe Region wie für Ihre Managed Airflow-Umgebung zu verwenden.

    gcloud compute networks subnets update default \
    --region DATAPROC_SERVERLESS_REGION \
    --enable-private-ip-google-access

  1. Erstellen Sie ein Speicherkonto mit den Standardeinstellungen.

  2. Rufen Sie den Zugriffsschlüssel und die Verbindungsstring ab für Ihr Speicherkonto.

  3. Erstellen Sie in Ihrem neu erstellten Speicherkonto einen Container mit den Standardoptionen.

  4. Erteilen Sie die Rolle „Storage Blob Delegator“ für den im vorherigen Schritt erstellten Container.

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

  6. Erstellen Sie ein SAS-Token für den Blockblob, den 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

Von Managed Airflow aus mit Azure verbinden

Fügen Sie Ihre Microsoft Azure Verbindung über die Airflow-Benutzeroberfläche hinzu:

  1. Rufen Sie Admin > Verbindungen auf.

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

    • Verbindungs-ID:azure_blob_connection
    • Verbindungstyp:Azure Blob Storage
    • Blob Storage-Anmeldung:Name Ihres Speicherkontos
    • Blob Storage-Schlüssel:Zugriffsschlüssel für Ihr Speicherkonto
    • Verbindungsstring für Blob Storage-Konto:Verbindungsstring für Ihr Speicherkonto
    • SAS-Token:SAS-Token, das aus Ihrem Blob generiert wurde

Datenverarbeitung mit Managed Service for Apache Spark

Beispiel für einen PySpark-Job

Der folgende 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.

DAG für die Datenanalyse

Beispiel für einen DAG

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

  • Mit dem AzureBlobStorageToGCSOperator wird die Datei holidays.csv von Ihrem Azure-Blockblob in Ihren Cloud Storage-Bucket übertragen.

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

  • The DataprocCreateBatchOperator erstellt und führt einen PySpark-Batchjob mit Managed Service for Apache Spark aus.

  • Mit dem BigQueryInsertJobOperator werden die Daten aus holidays.csv in der "Date" Spalte mit den Wetterdaten aus dem öffentlichen BigQuery-Dataset ghcn_d zusammengeführt. Die BigQueryInsertJobOperator Aufgaben werden dynamisch mit einer for-Schleife generiert und befinden sich in einer TaskGroup um die Lesbarkeit in der Diagrammansicht 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.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, Variablen sind eine universelle Möglichkeit, beliebige Einstellungen oder Konfigurationen als einfachen Schlüssel-Wert-Speicher zu speichern und abzurufen. In diesem DAG werden Airflow-Variablen verwendet, um allgemeine Werte zu speichern. So fügen Sie sie Ihrer Umgebung hinzu:

  1. Rufen Sie die Airflow-Benutzeroberfläche über die Managed Airflow-Konsole auf.

  2. Rufen Sie Admin > Variablen auf.

  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 Managed Service for Apache Spark-Job ausführen möchten, der die Netzwerkanforderungen von Managed Service for Apache Spark erfüllt. Dies ist die Region, in der Sie zuvor den privaten Google-Zugriff aktiviert haben.

    • dataproc_service_account: Das Dienstkonto für Ihre Managed Airflow-Umgebung. Sie finden dieses Dienstkonto auf dem Tab „Umgebungskonfiguration“ für Ihre Managed Airflow-Umgebung.

    • azure_blob_name: Der Name des Blobs, den Sie zuvor erstellt haben.

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

DAG in den Bucket Ihrer Umgebung hochladen

Managed Airflow 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 azureblobstoretogcsoperator_tutorial.py auf Ihrem lokalen Computer.

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

    Zu den 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 Managed Airflow-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 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. Die Zahlen in der Spalte „Wert“ sind in Zehntelgrad Celsius angegeben.

  5. Klicken Sie auf holidays_weather_normalized.

  6. Klicken Sie auf „Vorschau“, um die resultierende Tabelle anzusehen. 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:

Nächste Schritte