Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1
Questo tutorial mostra come utilizzare Cloud Composer per creare un DAG di Apache Airflow. Il DAG unisce i dati di un set di dati pubblico BigQuery e di un file CSV archiviato in un bucket Cloud Storage, quindi esegue un job batch Dataproc Serverless per elaborare i dati uniti.
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à degli Stati Uniti dal 1997 al 2021.
La domanda a cui vogliamo rispondere utilizzando il DAG è: "Quanto faceva caldo a Chicago in occasione del Giorno del Ringraziamento negli ultimi 25 anni?"
Obiettivi
- Creare un ambiente Cloud Composer nella configurazione predefinita
- crea un set di dati BigQuery vuoto
- Creare un nuovo bucket Cloud Storage
- Crea ed esegui un DAG che includa le seguenti attività:
- Carica un set di dati esterno da Cloud Storage in BigQuery
- Unisci 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 e dei bucket di ambienti Cloud Composer.
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 i parametri predefinite:
- Scegli una regione con sede negli Stati Uniti.
- Scegli l'ultima versione di Cloud Composer.
Concedi i seguenti ruoli all'account di servizio utilizzato nel tuo ambiente Cloud Composer affinché i worker Airflow possano eseguire correttamente le attività DAG:
- Utente BigQuery (
roles/bigquery.user
) - Proprietario dei dati BigQuery (
roles/bigquery.dataOwner
) - Utente account di servizio (
roles/iam.serviceAccountUser
) - Editor Dataproc (
roles/dataproc.editor
) - Dataproc Worker (
roles/dataproc.worker
)
- Utente BigQuery (
Creare risorse correlate
Crea un set di dati BigQuery vuoto con i seguenti parametri:
- Nome:
holiday_weather
- Regione:
US
- Nome:
Crea un nuovo bucket Cloud Storage nella regione multipla
US
.Esegui il seguente 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
Elaborazione dei dati utilizzando Dataproc Serverless
Esplora il job PySpark di esempio
Il codice riportato di seguito è un esempio di job PySpark che converte la temperatura da décimos de grado Celsius a gradi Celsius. Questo job converte i dati sulla temperatura del set di dati in un formato diverso.
Carica i file di supporto su Cloud Storage
Per caricare il file PySpark e il set di dati memorizzato in holidays.csv
:
Salva data_analytics_process.py sulla tua macchina locale.
Salva holidays.csv sulla tua macchina locale.
Nella console Google Cloud, vai alla pagina Browser Cloud Storage:
Fai clic sul nome del bucket creato in precedenza.
Nella scheda Oggetti del bucket, fai clic sul pulsante Carica file, seleziona
data_analytics_process.py
eholidays.csv
nella finestra di dialogo 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:
GCSToBigQueryOperator
importa il file holidays.csv da Cloud Storage in una nuova tabella del set di dati BigQueryholidays_weather
creato in precedenza.Il script
DataprocCreateBatchOperator
crea ed esegue un job batch PySpark utilizzando Dataproc Serverless.BigQueryInsertJobOperator
unisce i dati di holidays.csv nella colonna "Data" con i dati meteo del set di dati pubblico BigQuery ghcn_d. Le attivitàBigQueryInsertJobOperator
vengono generate dinamicamente utilizzando un ciclo for e si trovano in unTaskGroup
per una migliore leggibilità nella visualizzazione del grafico dell'interfaccia utente di Airflow.
Utilizzare l'interfaccia utente 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 di chiavi e valori. Questo DAG utilizza le variabili Airflow per memorizzare i valori comuni. Per aggiungerle al tuo ambiente:
Accedi all'interfaccia utente di Airflow dalla console Cloud Composer.
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 rete di Dataproc Serverless. Si tratta della regione in cui hai attivato l'accesso privato Google in precedenza.dataproc_service_account
: l'account di servizio per il tuo ambiente Cloud Composer. Puoi trovare questo account servizio nella scheda di configurazione dell'ambiente per il tuo ambiente Cloud Composer.
Carica il DAG nel bucket del tuo ambiente
Cloud Composer pianifica i DAG che si trovano nella
/dags
cartella del bucket del tuo ambiente. Per caricare il DAG utilizzando la console Google Cloud:
Sulla tua macchina locale, salva data_analytics_dag.py.
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, nella colonna Cartella DAG, fai clic sul link DAG. Viene aperta la cartella DAG del tuo ambiente.
Fai clic su Carica file.
Seleziona
data_analytics_dag.py
sul computer locale e fai clic su Apri.
Attiva il DAG
Nell'ambiente Cloud Composer, fai clic sulla scheda DAG.
Fai clic sull'ID DAG
data_analytics_dag
.Fai clic su Attiva DAG.
Attendi circa cinque-dieci 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 valore 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 valore sono in gradi Celsius.
Approfondimento con Dataproc Serverless (facoltativo)
Puoi provare una versione avanzata di questo DAG con un flusso di elaborazione dei dati PySpark più complesso. Consulta l'estensione Dataproc per l'esempio di analisi dei dati su GitHub.
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, inclusa l'eliminazione manuale del bucket dell'ambiente.
Passaggi successivi
- Esegui un DAG di analisi dei dati in Google Cloud Utilizzare i dati di AWS.
- Esegui un DAG di Data Analytics in Azure.