Data Analytics-DAG in Google Cloud ausführen

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

In dieser Anleitung wird gezeigt, wie Sie mit Managed Airflow einen Apache Airflow-DAG erstellen. Im DAG werden Daten aus einem öffentlichen BigQuery-Dataset und einer in einem Cloud Storage-Bucket gespeicherten CSV-Datei zusammengeführt. Anschließend wird ein Batchjob für 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 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 an Thanksgiving in den letzten 25 Jahren?“

Hinweis :In dieser Anleitung wird Airflow 2 verwendet.

Ziele

• Managed Airflow-Umgebung in der Standardkonfiguration erstellen
• Leeres BigQuery-Dataset erstellen
• Neuen Cloud Storage-Bucket erstellen
• Erstellen und führen Sie eine DAG mit den folgenden Aufgaben aus:
  • 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.

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

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. Weitere Informationen zum Zuweisen von Rollen

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

Berechtigungen erteilen

Weisen Sie Ihrem Nutzerkonto die folgenden Rollen und Berechtigungen zu:

• Rollen zum Verwalten von Managed Airflow-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.

Managed Airflow-Umgebung erstellen und vorbereiten

1. Managed Airflow-Umgebung mit Standardparametern erstellen:

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

2. Weisen Sie dem Dienstkonto, das in Ihrer verwalteten Airflow-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 erstellen

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

   • Name: holiday_weather
   • Region: US

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

3. Führen Sie den folgenden Befehl aus, um privaten Google-Zugriff im Standardsubnetz in der Region zu aktivieren, in der Sie Managed Service for Apache Spark ausführen möchten, um die Netzwerkanforderungen zu erfüllen. Es wird empfohlen, dieselbe Region wie für Ihre verwaltete Airflow-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 US-Region befinden, um die Anforderungen an die Colocation für die Übertragung des öffentlichen BigQuery-Datasets ghcn_d zu erfüllen.

Datenverarbeitung mit Managed Service for Apache Spark

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.

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

Unterstützende Dateien in Cloud Storage hochladen

So laden Sie die PySpark-Datei und das Dataset hoch, die in holidays.csv gespeichert sind:

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

2. Speichern Sie holidays.csv auf Ihrem lokalen Computer.

3. Wechseln Sie in der Google Cloud Console zur Seite Cloud Storage-Browser:

   Zum Cloud Storage Browser

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

5. 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 und holidays.csv aus und klicken Sie auf Öffnen.

DAG für die Datenanalyse

Beispiel-DAG ansehen

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

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

• Mit dem Befehl DataprocCreateBatchOperator wird ein PySpark-Batchjob mit Managed Service for Apache Spark 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. Die BigQueryInsertJobOperator-Aufgaben werden dynamisch mit einer for-Schleife generiert und befinden sich in einem TaskGroup, um die Lesbarkeit in der Diagrammansicht der Airflow-Benutzeroberfläche zu verbessern.

Wichtig :Die DataprocCreateBatchOperator ist in den apache-airflow-providers-google-Paketversionen 6.2.0 und höher verfügbar. Die erforderlichen Versionen dieses Pakets sind in Ihrer Umgebung ab Managed Airflow-Version 1.17.10 und 2.0.3 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.utils.task_group import TaskGroup

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

# 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}}"
PROCESSING_PYTHON_FILE = f"gs://{BUCKET_NAME}/data_analytics_process.py"

BATCH_ID = "data-processing-{{ ts_nodash | lower}}"  # Dataproc serverless only allows lowercase characters
BATCH_CONFIG = {
    "runtime_config": {"version": "1.1"},
    "pyspark_batch": {
        "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(
    "data_analytics_dag",
    # Continue to run DAG once per day
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:
    create_batch = dataproc.DataprocCreateBatchOperator(
        task_id="create_batch",
        project_id=PROJECT_NAME,
        region="{{ var.value.gce_region }}",
        batch=BATCH_CONFIG,
        batch_id=BATCH_ID,
    )
    # This data is static and it is safe to use WRITE_TRUNCATE
    # to reduce chance of 409 duplicate errors
    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):
            # BigQuery configs
            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",
            )

        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. In diesem DAG werden Airflow-Variablen zum Speichern gemeinsamer Werte verwendet. So fügen Sie sie Ihrer Umgebung hinzu:

1. Über die Managed Airflow-Konsole auf die Airflow-UI zugreifen

2. Klicken Sie auf Verwaltung > 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 Managed Service for Apache Spark-Job ausführen möchten, der den Netzwerkanforderungen für Managed Service for Apache Spark entspricht. Das ist die Region, in der Sie den privaten Google-Zugriff zuvor aktiviert haben.

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

DAG in den Bucket Ihrer Umgebung hochladen

Managed Airflow plant DAGs, die sich im /dags-Ordner des Buckets Ihrer Umgebung befinden. So laden Sie den DAG über dieGoogle Cloud Console hoch:

1. Speichern Sie data_analytics_dag.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 data_analytics_dag.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 data_analytics_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 angibt, 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 Explorer-Bereich auf den Namen Ihres Projekts.

3. Klicken Sie auf holidays_weather_joined.

4. Klicken Sie auf „Vorschau“, um die resultierende Tabelle aufzurufen. Die Zahlen in der Spalte „Wert“ werden in Zehntelgrad Celsius angegeben.

5. Klicken Sie auf holidays_weather_normalized.

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

Managed Service for Apache Spark im Detail (optional)

Sie können eine erweiterte Version dieses DAG mit einem komplexeren PySpark-Datenverarbeitungsablauf ausprobieren. Weitere Informationen finden Sie unter Managed Service for Apache Spark extension for the Data Analytics Example auf GitHub.

Bereinigen

Löschen Sie die einzelnen 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 das BigQuery-Dataset.

• Löschen Sie die verwaltete Airflow-Umgebung, einschließlich des manuellen Löschens des Buckets der Umgebung.

Nächste Schritte

• Datenanalyse-DAG in Google Cloud Mit Daten von AWS ausführen
• DAG für die Datenanalyse in Azure ausführen: