Esegui un DAG di analisi dei dati in Google Cloud utilizzando i dati di Azure

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

Questo tutorial è una modifica di Esegui un DAG di analisi dei dati in Google Cloud che mostra come connettere l'ambiente Managed Airflow a Microsoft Azure per utilizzare i dati archiviati. Mostra come utilizzare Managed Airflow per creare un DAG Apache Airflow. Il DAG unisce i dati di un set di dati pubblici BigQuery e di un file CSV archiviato in un Azure Blob Storage e poi esegue un job batch Managed Service for Apache Spark per elaborare i dati unificati.

Il set di dati pubblico di BigQuery in questo tutorial è ghcn_d, un database integrato di riepiloghi climatici in tutto il mondo. Il file CSV contiene informazioni sulle date e sui nomi delle festività statunitensi dal 1997 al 2021.

La domanda a cui vogliamo rispondere utilizzando il DAG è: "Che temperatura c'era a Chicago il giorno del Ringraziamento negli ultimi 25 anni?"

Obiettivi

  • Crea un ambiente Managed Airflow nella configurazione predefinita
  • Crea un blob in Azure
  • crea un set di dati BigQuery vuoto
  • Crea un nuovo bucket Cloud Storage
  • Crea ed esegui un DAG che includa le seguenti attività:
    • Caricare un set di dati esterno da Azure Blob Storage a Cloud Storage
    • Carica un set di dati esterno da Cloud Storage a BigQuery
    • Unire due set di dati in BigQuery
    • Esegui un job PySpark di analisi dei dati

Prima di iniziare

Abilita API

Abilita le seguenti API:

Console

Abilita le API Managed Service for Apache Spark, Managed Airflow, BigQuery e Cloud Storage.

Ruoli richiesti per abilitare le API

Per abilitare le API, devi disporre del ruolo IAM Amministratore utilizzo dei servizi (roles/serviceusage.serviceUsageAdmin), che include l'autorizzazione serviceusage.services.enable. Scopri come concedere i ruoli.

Abilita le API

gcloud

Abilita le API Managed Service for Apache Spark, Managed Airflow, BigQuery e Cloud Storage:

Ruoli richiesti per abilitare le API

Per abilitare le API, devi disporre del ruolo IAM Amministratore utilizzo dei servizi (roles/serviceusage.serviceUsageAdmin), che include l'autorizzazione serviceusage.services.enable. Scopri come concedere i ruoli. 

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

Concedi le autorizzazioni

Concedi i seguenti ruoli e autorizzazioni al tuo account utente:

Crea e prepara l'ambiente Managed Airflow

  1. Crea un ambiente Managed Airflow con parametri predefiniti:

  2. Concedi i seguenti ruoli al account di servizio utilizzato nel tuo ambiente Managed Airflow affinché i worker Airflow eseguano correttamente le attività DAG:

    • Utente BigQuery (roles/bigquery.user)
    • Proprietario dati BigQuery (roles/bigquery.dataOwner)
    • Service Account User (roles/iam.serviceAccountUser)
    • Editor Dataproc (roles/dataproc.editor)
    • Dataproc Worker (roles/dataproc.worker)

  1. Installa il apache-airflow-providers-microsoft-azure pacchetto PyPI nel tuo ambiente Managed Airflow.

  2. Crea un set di dati BigQuery vuoto con i seguenti parametri:

    • Nome: holiday_weather
    • Regione: US

  3. Crea un nuovo bucket Cloud Storage nella multiregione US.

  4. Esegui questo comando per abilitare l'accesso privato Google nella subnet predefinita della regione in cui vuoi eseguire Managed Service for Apache Spark per soddisfare i requisiti di rete. Ti consigliamo di utilizzare la stessa regione del tuo ambiente Managed Airflow.

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

  1. Crea un account di archiviazione con le impostazioni predefinite.

  2. Ottieni la chiave di accesso e la stringa di connessione per il tuo account di archiviazione.

  3. Crea un contenitore con le opzioni predefinite nel tuo account di archiviazione appena creato.

  4. Concedi il ruolo Delegatore di blob di archiviazione per il container creato nel passaggio precedente.

  5. Carica holidays.csv per creare un BLOB di blocchi con le opzioni predefinite nel portale Azure.

  6. Crea un token SAS per il blob di blocchi creato nel passaggio precedente nel portale Azure.

    • Metodo di firma: chiave di delega utente
    • Autorizzazioni: lettura
    • Indirizzo IP consentito: nessuno
    • Protocolli consentiti: solo HTTPS

Connettersi ad Azure da Managed Airflow

Aggiungi la tua connessione Microsoft Azure utilizzando la UI di Airflow:

  1. Vai ad Amministrazione > Connessioni.

  2. Crea una nuova connessione con la seguente configurazione:

    • ID connessione: azure_blob_connection
    • Tipo di connessione: Azure Blob Storage
    • Accesso a Blob Storage:il nome del tuo account di archiviazione
    • Chiave Blob Storage:la chiave di accesso per il tuo account di archiviazione
    • Stringa di connessione dell'account Blob Storage:la stringa di connessione dell'account di archiviazione
    • Token SAS:il token SAS generato dal blob

Elaborazione dei dati utilizzando Managed Service for Apache Spark

Esplora il job PySpark di esempio

Il codice mostrato di seguito è un esempio di job PySpark che converte la temperatura da decimi di grado Celsius a gradi Celsius. Questo job converte i dati di temperatura del set di dati in un formato diverso.

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

Carica il file PySpark su Cloud Storage

Per caricare il file PySpark su Cloud Storage:

  1. Salva data_analytics_process.py sulla tua macchina locale.

  2. Nella console Google Cloud , vai alla pagina Browser Cloud Storage:

    Vai al browser Cloud Storage

  3. Fai clic sul nome del bucket che hai creato in precedenza.

  4. Nella scheda Oggetti del bucket, fai clic sul pulsante Carica i file, seleziona data_analytics_process.py nella finestra di dialogo che viene visualizzata e fai clic su Apri.

DAG di analisi dei dati

Esplora il DAG di esempio

Il DAG utilizza più operatori per trasformare e unificare i dati:

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

Utilizzare la UI di Airflow per aggiungere variabili

In Airflow, le variabili sono un modo universale per archiviare e recuperare impostazioni o configurazioni arbitrarie come un semplice archivio chiave-valore. Questo DAG utilizza le variabili Airflow per memorizzare i valori comuni. Per aggiungerli al tuo ambiente:

  1. Accedi all'interfaccia utente di Airflow dalla console di Managed Airflow.

  2. Vai ad Amministrazione > Variabili.

  3. Aggiungi le seguenti variabili:

    • gcp_project: il tuo ID progetto.

    • gcs_bucket: il nome del bucket creato in precedenza (senza il prefisso gs://).

    • gce_region: la regione in cui vuoi che il job Managed Service for Apache Spark soddisfi i requisiti di networking di Managed Service for Apache Spark. Questa è la regione in cui hai attivato l'accesso privato Google in precedenza.

    • dataproc_service_account: il account di servizio per il tuo ambiente Managed Airflow. Puoi trovare questo service account nella scheda di configurazione dell'ambiente per il tuo ambiente Managed Airflow.

    • azure_blob_name: il nome del blob creato in precedenza.

    • azure_container_name: il nome del container creato in precedenza.

Carica il DAG nel bucket del tuo ambiente

Managed Airflow pianifica i DAG che si trovano nella cartella /dags del bucket del tuo ambiente. Per caricare il DAG utilizzando la consoleGoogle Cloud :

  1. Sulla tua macchina locale, salva azureblobstoretogcsoperator_tutorial.py.

  2. Nella console Google Cloud , vai alla pagina Ambienti.

    Vai ad Ambienti

  3. Nell'elenco degli ambienti, nella colonna Cartella DAG, fai clic sul link DAG. Si apre la cartella DAG del tuo ambiente.

  4. Fai clic su Carica file.

  5. Seleziona azureblobstoretogcsoperator_tutorial.py sulla macchina locale e fai clic su Apri.

Attiva il DAG

  1. Nell'ambiente Managed Airflow, fai clic sulla scheda DAG.

  2. Fai clic sull'ID DAG azure_blob_to_gcs_dag.

  3. Fai clic su Attiva DAG.

  4. Attendi circa 5-10 minuti finché non viene visualizzato un segno di spunta verde che indica che le attività sono state completate correttamente.

Convalida l'esito positivo del DAG

  1. Nella console Google Cloud , vai alla pagina BigQuery.

    Vai a BigQuery

  2. Nel riquadro Explorer, fai clic sul nome del progetto.

  3. Fai clic su holidays_weather_joined.

  4. Fai clic su Anteprima per visualizzare la tabella risultante. Tieni presente che i numeri nella colonna dei valori sono in decimi di grado Celsius.

  5. Fai clic su holidays_weather_normalized.

  6. Fai clic su Anteprima per visualizzare la tabella risultante. Tieni presente che i numeri nella colonna dei valori sono in gradi Celsius.

Esegui la pulizia

Elimina le singole risorse che hai creato per questo tutorial:

Passaggi successivi