Panoramica

I workflow sono un tema molto comune nell'analisi dei dati: implicano l'acquisizione, la trasformazione e l'analisi dei dati per individuare le informazioni significative al loro interno. In Google Cloud, per ospitare i workflow viene usato Cloud Composer, una versione in hosting del popolare strumento per workflow open source Apache Airflow.

In questo lab utilizzerai la console Google Cloud per configurare un ambiente Cloud Composer. Poi userai Cloud Composer per eseguire un semplice workflow che verifica l'esistenza di un file di dati, crea un cluster Cloud Dataproc, esegue un job di conteggio parole di Apache Hadoop nel cluster Cloud Dataproc ed elimina il cluster.

In questo lab proverai a:

• Usare la console Google Cloud per creare l'ambiente Cloud Composer

• Visualizzare ed eseguire il DAG (grafo diretto aciclico) nell'interfaccia web di Airflow

• Visualizzare i risultati del job di conteggio parole in Storage

Configurazione e requisiti

Configurazione del lab

Per ciascun lab, riceverai un nuovo progetto Google Cloud e un insieme di risorse per un periodo di tempo limitato senza alcun costo aggiuntivo.

1. Accedi a Qwiklabs utilizzando una finestra di navigazione in incognito.

2. Tieni presente la durata dell'accesso al lab (ad esempio, 1:15:00) e assicurati di finire entro quell'intervallo di tempo.
   Non è disponibile una funzionalità di pausa. Se necessario, puoi riavviare il lab ma dovrai ricominciare dall'inizio.

3. Quando è tutto pronto, fai clic su Inizia lab.

4. Annota le tue credenziali del lab (Nome utente e Password). Le userai per accedere a Google Cloud Console.

5. Fai clic su Apri console Google.

6. Fai clic su Utilizza un altro account e copia/incolla le credenziali per questo lab nei prompt.
   Se utilizzi altre credenziali, compariranno errori oppure ti verranno addebitati dei costi.

7. Accetta i termini e salta la pagina di ripristino delle risorse.

Attiva Google Cloud Shell

Google Cloud Shell è una macchina virtuale in cui sono caricati strumenti per sviluppatori. Offre una home directory permanente da 5 GB e viene eseguita su Google Cloud.

Google Cloud Shell fornisce l'accesso da riga di comando alle risorse Google Cloud.

1. Nella barra degli strumenti in alto a destra della console Cloud, fai clic sul pulsante Apri Cloud Shell.

   

2. Fai clic su Continua.

Bastano pochi istanti per eseguire il provisioning e connettersi all'ambiente. Quando la connessione è attiva, l'autenticazione è già avvenuta e il progetto è impostato sul tuo PROJECT_ID. Ad esempio:

gcloud è lo strumento a riga di comando di Google Cloud. È preinstallato su Cloud Shell e supporta il completamento.

• Puoi visualizzare il nome dell'account attivo con questo comando:

gcloud auth list

Output:

Credentialed accounts: - @.com (active)

Output di esempio:

Credentialed accounts: - google1623327_student@qwiklabs.net

• Puoi elencare l'ID progetto con questo comando:

gcloud config list project

Output:

[core] project =

Output di esempio:

[core] project = qwiklabs-gcp-44776a13dea667a6 Nota: la documentazione completa di gcloud è disponibile nella guida Panoramica dell'interfaccia a riga di comando gcloud .

Verifica le autorizzazioni del progetto

Prima di iniziare il tuo lavoro su Google Cloud, devi assicurarti che il tuo progetto disponga delle autorizzazioni corrette in Identity and Access Management (IAM).

1. Nella console Google Cloud, nel menu di navigazione (), seleziona IAM e amministrazione > IAM.

2. Conferma che l'account di servizio di computing predefinito {project-number}-compute@developer.gserviceaccount.com sia presente e che abbia il ruolo di editor assegnato. Il prefisso dell'account è il numero del progetto, che puoi trovare in Menu di navigazione > Panoramica di Cloud > Dashboard

Nota: se l'account non è presente in IAM o non dispone del ruolo editor, attieniti alla procedura riportata di seguito per assegnare il ruolo richiesto.

1. Nel menu di navigazione della console Google Cloud, fai clic su Panoramica di Cloud > Dashboard.
2. Copia il numero del progetto (es. 729328892908).
3. Nel menu di navigazione, seleziona IAM e amministrazione > IAM.
4. Nella parte superiore della tabella dei ruoli, sotto Visualizza per entità, fai clic su Concedi accesso.
5. Per Nuove entità, digita:

{project-number}-compute@developer.gserviceaccount.com

6. Sostituisci {project-number} con il numero del tuo progetto.
7. Come Ruolo, seleziona Progetto (o Base) > Editor.
8. Fai clic su Salva.

Attività 1: attività di ambiente

1. Nella barra del titolo della console Google Cloud, fai clic su Attiva Cloud Shell. Se richiesto, fai clic su Continua.

2. Esegui questi comandi per assegnare il ruolo Composer Worker all'account di servizio per sviluppatori di Compute.

export PROJECT_ID=$(gcloud config get-value project) export PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format="value(projectNumber)") gcloud projects add-iam-policy-binding {{{project_0.project_id}}} \ --member=serviceAccount:$PROJECT_NUMBER-compute@developer.gserviceaccount.com \ --role=roles/composer.worker

3. Esegui questi comandi per assicurarti che le API richieste siano abilitate in modo corretto nel tuo progetto.

gcloud services disable composer.googleapis.com gcloud services disable artifactregistry.googleapis.com gcloud services disable container.googleapis.com gcloud services enable artifactregistry.googleapis.com gcloud services enable container.googleapis.com gcloud services enable composer.googleapis.com

Attività 2: crea l'ambiente Cloud Composer

In questa sezione, creerai l'ambiente Cloud Composer.

Nota: prima di andare avanti, assicurati di aver eseguito le attività precedenti per garantire che le API richieste siano abilitate correttamente. Se non l'hai ancora fatto, esegui queste attività ora, altrimenti la creazione dell'ambiente Cloud Composer non riuscirà.

1. Nella barra del titolo della console Google Cloud, digita Composer nel campo di ricerca, quindi fai clic su Composer nella sezione Prodotti e pagine.

2. Fai clic su Crea ambiente e seleziona Composer 3. Imposta quanto segue per l'ambiente:

Proprietà	Valore
Nome	highcpu
Località
Versione immagine	composer-3-airflow-n.n.n-build.n (Nota: seleziona l'immagine con il numero più alto disponibile)
Service account	xxxxx-compute@developer.gserviceaccount.com
Modalità di resilienza	Resilienza standard
Zona del database Airflow

3. In Risorse dell'ambiente, seleziona Piccolo.

Lascia invariate tutte le altre impostazioni predefinite.

4. Fai clic su Crea.

Il processo di creazione dell'ambiente è completato quando viene visualizzato il segno di spunta verde a sinistra del nome dell'ambiente nella pagina Ambienti della console.

Il completamento del processo di configurazione dell'ambiente può richiedere 15-30 minuti. Prosegui con il lab mentre l'ambiente viene avviato.

Fai clic su Controlla i miei progressi per verificare l'obiettivo.

Crea l'ambiente Cloud Composer.

Crea un bucket Cloud Storage

Crea un bucket Cloud Storage nel tuo progetto. Questo bucket verrà usato come output per il job Hadoop da Dataproc.

1. Vai a Menu di navigazione > Cloud Storage > Bucket e fai clic su + Crea.

2. Assegna al bucket un nome universalmente univoco, ad esempio l'ID progetto , poi fai clic su Crea. Se appare il messaggio L'accesso pubblico verrà vietato fai clic su Conferma.

Memorizza il nome del bucket Cloud Storage per utilizzarlo come variabile Airflow più avanti nel lab.

Fai clic su Controlla i miei progressi per verificare l'obiettivo.

Crea un bucket Cloud Storage.

Attività 3: Airflow e concetti principali

Mentre aspetti che l'ambiente Composer venga creato, rivediamo alcuni termini usati in Airflow.

Airflow è una piattaforma per creare, pianificare e monitorare i workflow in modo programmatico.

Usa Airflow per creare workflow come grafi diretti aciclici (DAG) di attività. Lo scheduler di Airflow esegue le attività su un array di worker in base alle dipendenze specificate.

Concetti principali

DAG

Un grafo diretto aciclico (Directed Acyclic Graph, DAG) è una raccolta di tutte le attività da eseguire, organizzate in modo da riflettere le rispettive relazioni e dipendenze.

Operatore

La descrizione di una singola attività, di solito a livello atomico. Ad esempio, BashOperator viene utilizzato per eseguire i comandi Bash.

Attività

Un'istanza con parametri di un operatore; un nodo nel DAG.

Istanza dell'attività

Un'esecuzione specifica di un'attività, caratterizzata come DAG, attività e point-in-time. Ha uno stato indicativo: running, success, failed, skipped…

Per saperne di più sui concetti di Airflow, consulta la documentazione.

Attività 4: definisci il workflow

Adesso parliamo del workflow che userai. I workflow di Cloud Composer sono composti da DAG (grafi diretti aciclici). I DAG sono definiti in file Python standard che si trovano nella cartella DAG_FOLDER di Airflow. Airflow esegue il codice in ogni file per creare in modo dinamico gli oggetti DAG. Puoi avere tutti i DAG che vuoi, ognuno dei quali descrive un numero qualsiasi di attività. In generale, ogni DAG deve corrispondere a un singolo workflow logico.

Di seguito è riportato il codice del workflow hadoop_tutorial.py, chiamato anche DAG:

"""Example Airflow DAG that creates a Cloud Dataproc cluster, runs the Hadoop wordcount example, and deletes the cluster. This DAG relies on three Airflow variables https://airflow.apache.org/concepts.html#variables * gcp_project - Google Cloud Project to use for the Cloud Dataproc cluster. * gce_zone - Google Compute Engine zone where Cloud Dataproc cluster should be created. * gce_region - Google Compute Engine region where Cloud Dataproc cluster should be created. * gcs_bucket - Google Cloud Storage bucket to used as output for the Hadoop jobs from Dataproc. See https://cloud.google.com/storage/docs/creating-buckets for creating a bucket. """ import datetime import os from airflow import models from airflow.providers.google.cloud.operators.dataproc import ( DataprocCreateClusterOperator, DataprocDeleteClusterOperator, DataprocSubmitJobOperator, ) from airflow.utils.trigger_rule import TriggerRule # Define the DAG's schedule using a cron expression for daily runs DAILY_SCHEDULE = '0 0 * * *' # Runs at midnight every day # Output file for Cloud Dataproc job. output_file = os.path.join( models.Variable.get('gcs_bucket'), 'wordcount', datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep # Path to Hadoop wordcount example available on every Dataproc cluster. WORDCOUNT_JAR = ( 'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar' ) # Arguments to pass to Cloud Dataproc job. wordcount_args = ['wordcount', 'gs://pub/shakespeare/rose.txt', output_file] 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, # If a task fails, retry it once after waiting at least 5 minutes 'retries': 1, 'retry_delay': datetime.timedelta(minutes=5), 'project_id': models.Variable.get('gcp_project') } with models.DAG( dag_id='composer_hadoop_tutorial', # It's good practice to explicitly use 'dag_id' schedule_interval=DAILY_SCHEDULE, # Using the new cron expression default_args=default_dag_args) as dag: # Create a Cloud Dataproc cluster. create_dataproc_cluster = DataprocCreateClusterOperator( task_id='create_dataproc_cluster', # Give the cluster a unique name by appending the date scheduled. # See https://airflow.apache.org/code.html#default-variables cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}', num_workers=2, region=models.Variable.get('gce_region'), zone=models.Variable.get('gce_zone'), image_version='2.0', master_machine_type='e2-standard-2', worker_machine_type='e2-standard-2') # Run the Hadoop wordcount example installed on the Cloud Dataproc cluster # master node. run_dataproc_hadoop = DataprocSubmitJobOperator( task_id='run_dataproc_hadoop', region=models.Variable.get('gce_region'), job={ "placement": {"cluster_name": 'composer-hadoop-tutorial-cluster-{{ ds_nodash }}'}, "hadoop_job": { "main_jar_file_uri": WORDCOUNT_JAR, "args": wordcount_args } } ) # Delete Cloud Dataproc cluster. delete_dataproc_cluster = DataprocDeleteClusterOperator( task_id='delete_dataproc_cluster', project_id=models.Variable.get('gcp_project'), region=models.Variable.get('gce_region'), cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}', # Setting trigger_rule to ALL_DONE causes the cluster to be deleted # even if the Dataproc job fails. trigger_rule=TriggerRule.ALL_DONE) # Define DAG dependencies. create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

Per orchestrare le tre attività del workflow, il DAG importa i seguenti operatori:

1. DataprocCreateClusterOperator: crea un cluster Cloud Dataproc.
2. DataprocSubmitJobOperator: invia un job di conteggio parole di Hadoop e scrive i risultati in un bucket Cloud Storage.
3. DataprocDeleteClusterOperator: elimina il cluster per evitare addebiti continui per Compute Engine.

Le attività vengono eseguite in sequenza, come puoi vedere in questa sezione del file:

# Define DAG dependencies. create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

Il nome del DAG è composer_hadoop_tutorial e il DAG viene eseguito una volta al giorno:

with models.DAG( 'composer_hadoop_tutorial', # Continue to run DAG once per day schedule_interval=datetime.timedelta(days=1), default_args=default_dag_args) as dag:

Poiché la data di inizio specificata da start_date e passata a default_dag_args è impostato su yesterday, Cloud Composer pianifica il workflow in modo da eseguirlo subito dopo il caricamento del DAG.

Attività 5: visualizza le informazioni dell'ambiente

1. Torna a Composer per controllare lo stato dell'ambiente.

2. Una volta creato l'ambiente, fai clic sul suo nome (highcpu) per visualizzarne i dettagli.

Nella scheda Configurazione dell'ambiente vedrai informazioni come l'URL della UI web di Airflow, il cluster GKE e un link alla cartella DAG, che è archiviata nel tuo bucket.

Nota: Cloud Composer pianifica i workflow solo nella cartella /dags.

Attività 6: usa la UI di Airflow

Per accedere all'interfaccia web di Airflow mediante la console:

1. Torna alla pagina Ambienti.
2. Nella colonna Server web Airflow per l'ambiente, fai clic su Airflow.
3. Fai clic sulle credenziali del lab.
4. L'interfaccia web di Airflow si apre in una nuova finestra del browser.

Attività 7: imposta le variabili Airflow

Le variabili Airflow sono un concetto specifico di questa piattaforma e sono diverse dalle variabili di ambiente.

1. Nell'interfaccia di Airflow, seleziona Admin (Amministratore) > Variables (Variabili) dalla barra dei menu.

2. Fai clic sull'icona + per aggiungere un nuovo record.

3. Crea le seguenti variabili Airflow: gcp_project, gcs_bucket, gce_zone e gce_region. Fai clic su Save (Salva) dopo ogni variabile.

Key (Chiave)	Val (Valore)	Details (Dettagli)
gcp_project		Il progetto Google Cloud utilizzato in questo lab.
gcs_bucket	gs://	In questo bucket viene archiviato l'output dei job Hadoop da Dataproc.
gce_zone		Questa è la zona Compute Engine in cui verrà creato il cluster Cloud Dataproc.
gce_region		Questa è la regione Compute Engine in cui verrà creato il cluster Cloud Dataproc.

Fai clic su Save (Salva). Dopo aver aggiunto la prima variabile, ripeti la stessa procedura per la seconda e la terza variabile. Al termine, questo dovrebbe essere l'aspetto della tabella delle variabili:

Attività 8: carica il DAG in Cloud Storage

Per caricare il DAG:

1. In Cloud Shell esegui il comando riportato di seguito per caricare una copia del file hadoop_tutorial.py nel bucket Cloud Storage generato automaticamente quando hai creato l'ambiente.

2. Sostituisci <DAGs_folder_path> con il percorso della cartella dei DAG in questo comando:

gcloud storage cp gs://cloud-training/datawarehousing/lab_assets/hadoop_tutorial.py <DAGs_folder_path>

• Puoi recuperare il percorso in Composer.
• Fai clic sull'ambiente creato in precedenza e poi sulla scheda Configurazione dell'ambiente per vedere i dettagli dell'ambiente.
• Trova Cartella DAG e copia il percorso.

Il comando modificato per caricare il file sarà simile a questo:

gcloud storage cp gs://cloud-training/datawarehousing/lab_assets/hadoop_tutorial.py gs://{{{project_0.default_region|REGION}}}-highcpu-0682d8c0-bucket/dags

3. Dopo aver caricato il file nella directory dei DAG, apri la cartella dags nel bucket. Vedrai il file nella scheda Oggetti dei dettagli del bucket.

Quando un file DAG viene aggiunto alla cartella dei DAG, Cloud Composer aggiunge il DAG ad Airflow e lo pianifica automaticamente. Le modifiche ai DAG vengono applicate entro 3-5 minuti.

Puoi vedere lo stato dell'attività del DAG composer_hadoop_tutorial nell'interfaccia web di Airflow.

Nota: puoi ignorare senza problemi eventuali messaggi visualizzati nell'interfaccia, ad esempio "The scheduler does not appear to be running…" (Sembra che lo scheduler non sia in esecuzione). L'interfaccia web di Airflow viene aggiornata man mano che il DAG viene elaborato.

Fai clic su Controlla i miei progressi per verificare l'obiettivo.

Carica il DAG in Cloud Storage.

Esplora le esecuzioni del DAG

Quando carichi il file DAG nella cartella dags di Cloud Storage, Cloud Composer analizza il file. Se non vengono rilevati errori, il nome del workflow viene visualizzato nell'elenco di DAG e il workflow viene messo in coda per essere eseguito immediatamente.

1. Assicurati di trovarti nella scheda DAG dell'interfaccia web di Airflow. Il completamento di questo processo richiede vari minuti. Aggiorna il browser per assicurarti di visualizzare le informazioni più recenti.

2. In Airflow, fai clic su composer_hadoop_tutorial per aprire la pagina dei dettagli del DAG. Questa pagina include diverse rappresentazioni delle attività e delle dipendenze del workflow.

3. Fai clic su Graph (Grafo) nella barra degli strumenti. Passa il mouse sopra il grafo di ogni attività per visualizzarne lo stato. Puoi notare che il bordo attorno a ciascuna attività indica anche lo stato (bordo verde = in esecuzione; rosso = non riuscita e così via).

4. Fai clic sul link "Refresh" (Aggiorna) per assicurarti di visualizzare le informazioni più recenti. I bordi dei processi cambiano colore quando il rispettivo stato cambia.

Nota: se il cluster Dataproc esiste già, puoi eseguire nuovamente il workflow fino a quando non riesce facendo clic su "create_dataproc_cluster" e poi su Clear (Cancella) per reimpostare le tre attività. Fai clic su OK per confermare.

5. Quando lo stato per create_dataproc_cluster diventa "running", seleziona Menu di navigazione > Dataproc e fai clic su:

   • Cluster per monitorare la creazione e l'eliminazione del cluster. Il cluster creato dal workflow è temporaneo. Esiste solo per la durata del workflow e viene eliminato dall'ultima attività del workflow.
   • Job per monitorare il job di conteggio parole di Apache Hadoop. Fai clic sull'ID job per visualizzarne l'output del log.

6. Quando lo stato di Dataproc diventa "In esecuzione", torna ad Airflow e fai clic su Refresh (Aggiorna) per vedere che il cluster è completato.

Quando il processo run_dataproc_hadoop viene completato, seleziona Menu di navigazione > Cloud Storage > Bucket e fai clic sul nome del tuo bucket per vedere i risultati del conteggio parole nella cartella wordcount.

7. Una volta completati tutti i passaggi nel DAG, il bordo di ogni passaggio è verde scuro. Inoltre, il cluster Dataproc creato in precedenza è stato eliminato.

Complimenti!

Hai eseguito un workflow di Cloud Composer.

Passaggi successivi

• Guarda la presentazione di Cloud Composer al NEXT 18 di San Francisco: Flexible, Easy Data Pipelines on Google Cloud with Cloud Composer (Cloud Next '18)
• Per vedere il valore di una variabile, esegui il sottocomando dell'interfaccia a riga di comando di Airflow variables con l'argomento get oppure usa l'interfaccia web di Airflow.
• Per informazioni sull'interfaccia web di Airflow, consulta Accesso all'interfaccia web di Airflow.

Termina il lab

Una volta completato il lab, fai clic su Termina lab. Google Cloud Skills Boost rimuove le risorse che hai utilizzato ed esegue la pulizia dell'account.

Avrai la possibilità di inserire una valutazione in merito alla tua esperienza. Seleziona il numero di stelle applicabile, inserisci un commento, quindi fai clic su Invia.

Il numero di stelle corrisponde alle seguenti valutazioni:

• 1 stella = molto insoddisfatto
• 2 stelle = insoddisfatto
• 3 stelle = esperienza neutra
• 4 stelle = soddisfatto
• 5 stelle = molto soddisfatto

Se non vuoi lasciare un feedback, chiudi la finestra di dialogo.

Per feedback, suggerimenti o correzioni, utilizza la scheda Assistenza.

Copyright 2025 Google LLC Tutti i diritti riservati. Google e il logo Google sono marchi di Google LLC. Tutti gli altri nomi di società e prodotti sono marchi delle rispettive società a cui sono associati.