Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1
Questo tutorial è una modifica di Esegui un DAG di analisi dei dati in Google Cloud che mostra come connettere l'ambiente Cloud Composer a Microsoft Azure per utilizzare i dati archiviati. Mostra come utilizzare Cloud Composer per creare un DAG Apache Airflow. Il DAG unisce i dati di un set di dati pubblico BigQuery e di un file CSV archiviato in un Azure Blob Storage e poi esegue un job batch Dataproc Serverless 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 Cloud Composer nella configurazione predefinita
- Creare 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à:
- Carica un set di dati esterno da Azure Blob Storage a Cloud Storage
- Caricare un set di dati esterno da Cloud Storage in 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
Enable the Dataproc, Cloud Composer, BigQuery, Cloud Storage APIs.
gcloud
Enable the Dataproc, Cloud Composer, BigQuery, Cloud Storage APIs:
gcloud services enable dataproc.googleapis.comcomposer.googleapis.com bigquery.googleapis.com storage.googleapis.com
Concedi le autorizzazioni
Concedi i seguenti ruoli e autorizzazioni al tuo account utente:
Concedi ruoli per la gestione degli ambienti Cloud Composer e dei bucket degli ambienti.
Concedi il ruolo BigQuery Data Owner (
roles/bigquery.dataOwner
) per creare un set di dati BigQuery.Concedi il ruolo Amministratore Storage (
roles/storage.admin
) per creare un bucket Cloud Storage.
Crea e prepara l'ambiente Cloud Composer
Crea un ambiente Cloud Composer con parametri predefiniti:
- Scegli una regione con sede negli Stati Uniti.
- Scegli l'ultima versione di Cloud Composer.
Concedi i seguenti ruoli al account di servizio utilizzato nel tuo ambiente Cloud Composer affinché i worker Airflow eseguano correttamente le attività DAG:
- Utente BigQuery (
roles/bigquery.user
) - Proprietario dati BigQuery (
roles/bigquery.dataOwner
) - Utente service account (
roles/iam.serviceAccountUser
) - Editor Dataproc (
roles/dataproc.editor
) - Dataproc Worker (
roles/dataproc.worker
)
- Utente BigQuery (
Creare e modificare le risorse correlate in Google Cloud
Installa il
apache-airflow-providers-microsoft-azure
pacchetto PyPI nel tuo ambiente Cloud Composer.Crea un set di dati BigQuery vuoto con i seguenti parametri:
- Nome:
holiday_weather
- Regione:
US
- Nome:
Crea un nuovo bucket Cloud Storage nella multiregione
US
.Esegui questo comando per abilitare l'accesso privato Google nella subnet predefinita della regione in cui vuoi eseguire Dataproc Serverless per soddisfare i requisiti di rete. Ti consigliamo di utilizzare la stessa regione del tuo ambiente Cloud Composer.
gcloud compute networks subnets update default \ --region DATAPROC_SERVERLESS_REGION \ --enable-private-ip-google-access
Crea risorse correlate in Azure
Crea un account di archiviazione con le impostazioni predefinite.
Ottieni la chiave di accesso e la stringa di connessione per il tuo account di archiviazione.
Crea un contenitore con le opzioni predefinite nel tuo account di archiviazione appena creato.
Concedi il ruolo Delegatore di blob di archiviazione per il container creato nel passaggio precedente.
Carica holidays.csv per creare un blob di blocchi con le opzioni predefinite nel portale Azure.
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 Cloud Composer
Aggiungi la tua connessione Microsoft Azure utilizzando l'UI di Airflow:
Vai ad Amministrazione > Connessioni.
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
- ID connessione:
Elaborazione dei dati utilizzando Dataproc Serverless
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.
Carica il file PySpark su Cloud Storage
Per caricare il file PySpark in Cloud Storage:
Salva data_analytics_process.py sulla tua macchina locale.
Nella console Google Cloud , vai alla pagina Browser Cloud Storage:
Fai clic sul nome del bucket che hai creato in precedenza.
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 di dati
Esplora il DAG di esempio
Il DAG utilizza più operatori per trasformare e unificare i dati:
Il
AzureBlobStorageToGCSOperator
trasferisce il file holidays.csv dal blob di blocchi di Azure al bucket Cloud Storage.Il
GCSToBigQueryOperator
importa il file holidays.csv da Cloud Storage a una nuova tabella nel set di datiholidays_weather
BigQuery che hai creato in precedenza.DataprocCreateBatchOperator
crea ed esegue un job batch PySpark utilizzando Dataproc Serverless.BigQueryInsertJobOperator
unisce i dati di holidays.csv nella colonna "Date" con i dati meteo del set di dati pubblico BigQuery ghcn_d. IBigQueryInsertJobOperator
sono generati dinamicamente utilizzando un ciclo for e si trovano in unTaskGroup
per una migliore leggibilità nella visualizzazione a grafo della UI di Airflow.
Utilizzare l'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:
Vai ad Amministrazione > Variabili.
Aggiungi le seguenti variabili:
gcp_project
: il tuo ID progetto.gcs_bucket
: il nome del bucket creato in precedenza (senza il prefissogs://
).gce_region
: la regione in cui vuoi che il tuo job Dataproc soddisfi i requisiti di networking di Dataproc Serverless. Questa è la regione in cui hai abilitato l'accesso privato Google in precedenza.dataproc_service_account
: il account di servizio per il tuo ambiente Cloud Composer. Puoi trovare questo account di servizio nella scheda di configurazione dell'ambiente per il tuo ambiente Cloud Composer.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 dell'ambiente
Cloud Composer pianifica i DAG che si trovano nella cartella
/dags
del bucket dell'ambiente. Per caricare il DAG utilizzando la consoleGoogle Cloud :
Sulla tua macchina locale, salva azureblobstoretogcsoperator_tutorial.py.
Nella console Google Cloud , vai alla pagina Ambienti.
Nell'elenco degli ambienti, nella colonna Cartella DAG, fai clic sul link DAG. Si apre la cartella DAG del tuo ambiente.
Fai clic su Carica file.
Seleziona
azureblobstoretogcsoperator_tutorial.py
sulla macchina locale e fai clic su Apri.
Attiva il DAG
Nell'ambiente Cloud Composer, fai clic sulla scheda DAG.
Fai clic sull'ID DAG
azure_blob_to_gcs_dag
.Fai clic su Attiva DAG.
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
Nella console Google Cloud , vai alla pagina BigQuery.
Nel riquadro Explorer, fai clic sul nome del progetto.
Fai clic su
holidays_weather_joined
.Fai clic su Anteprima per visualizzare la tabella risultante. Tieni presente che i numeri nella colonna dei valori sono in decimi di grado Celsius.
Fai clic su
holidays_weather_normalized
.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:
Elimina il bucket Cloud Storage che hai creato per questo tutorial.
Elimina l'ambiente Cloud Composer, incluso l'eliminazione manuale del bucket dell'ambiente.