Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
Questa pagina descrive come utilizzare Cloud Composer 2 per eseguire carichi di lavoro Dataproc Serverless su Google Cloud.
Gli esempi nelle sezioni seguenti mostrano come utilizzare gli operatori per la gestione dei carichi di lavoro batch serverless di Dataproc. Puoi usare questi operatori nei DAG che creano, eliminano, elencano e recuperano un carico di lavoro batch Spark serverless di Dataproc:
Crea DAG per gli operatori che funzionano con i carichi di lavoro batch serverless di Dataproc:
Crea DAG che utilizzano container personalizzati e Dataproc Metastore.
Configura il server di cronologia permanente per questi DAG.
Prima di iniziare
Abilita l'API Dataproc:
Console
Attiva l'API Dataproc.
gcloud
Attiva l'API Dataproc.
gcloud services enable dataproc.googleapis.com
Seleziona la località per il file del carico di lavoro batch. Puoi utilizzare una delle seguenti opzioni:
- Crea un bucket Cloud Storage in cui archiviare questo file.
- Utilizza il bucket del tuo ambiente. Poiché non è necessario sincronizzare questo file
con Airflow, puoi creare una sottocartella separata all'esterno delle cartelle
/dags
o/data
. Ad esempio:/batches
. - Utilizza un bucket esistente.
configura file e variabili Airflow
Questa sezione illustra come impostare i file e configurare le variabili Airflow per questo tutorial.
Carica un file del carico di lavoro Spark ML serverless Dataproc in un bucket
Il carico di lavoro in questo tutorial esegue uno script pyspark:
Salva qualsiasi script pyspark in un file locale denominato
spark-job.py
. Ad esempio, puoi utilizzare lo script pyspark di esempio.Carica il file nella posizione selezionata in Prima di iniziare.
Imposta variabili Airflow
Gli esempi nelle sezioni seguenti utilizzano le variabili Airflow. Se imposti i valori per queste variabili in Airflow, il tuo codice DAG può accedere a questi valori.
Gli esempi in questo tutorial utilizzano le seguenti variabili Airflow. Puoi impostarle in base alle esigenze, a seconda dell'esempio che utilizzi.
Imposta le seguenti variabili Airflow da utilizzare nel codice DAG:
project_id
: ID progetto.bucket_name
: URI di un bucket in cui si trova il file Python principale del carico di lavoro (spark-job.py
). Hai selezionato questa località in Prima di iniziare.phs_cluster
: nome del cluster del server di cronologia permanente. La variabile viene impostata durante la creazione di un server di cronologia permanente.image_name
: nome e tag dell'immagine container personalizzata (image:tag
). Imposta questa variabile quando utilizzi un'immagine container personalizzata con DataprocCreateBatchOperator.metastore_cluster
: nome del servizio Dataproc Metastore. Imposta questa variabile quando utilizzi il servizio Dataproc Metastore con DataprocCreateBatchOperator.region_name
: regione in cui si trova il servizio Dataproc Metastore. Imposta questa variabile quando utilizzi il servizio Dataproc Metastore con DataprocCreateBatchOperator.
Usa la console Google Cloud e la UI di Airflow per impostare ogni variabile Airflow
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul link Airflow per il tuo ambiente. Si apre la UI di Airflow.
Nella UI di Airflow, seleziona Amministratore > Variabili.
Fai clic su Add a new record (Aggiungi un nuovo record).
Specifica il nome della variabile nel campo Chiave e imposta il relativo valore nel campo Valore.
Fai clic su Salva.
Crea un server di cronologia permanente
Utilizza un server di cronologia permanente (PHS) per visualizzare i file di cronologia di Spark dei carichi di lavoro batch:
- Crea un server di cronologia permanente.
- Assicurati di aver specificato il nome del cluster PHS nella
phs_cluster
variabile Airflow.
DataprocCreateBatchOperator
Il DAG seguente avvia un carico di lavoro batch serverless di Dataproc.
Per ulteriori informazioni sugli argomenti DataprocCreateBatchOperator
, consulta la pagina relativa al codice sorgente dell'operatore.
Per ulteriori informazioni sugli attributi che puoi passare nel parametro batch
di DataprocCreateBatchOperator
, consulta la
descrizione della classe Batch.
Usa l'immagine container personalizzata con DataprocCreateBatchOperator
L'esempio seguente mostra come utilizzare un'immagine container personalizzata per eseguire i carichi di lavoro. Puoi utilizzare un container personalizzato, ad esempio, per aggiungere dipendenze Python non fornite dall'immagine del container predefinito.
Per utilizzare un'immagine container personalizzata:
Crea un'immagine container personalizzata e caricala in Container Registry.
Specifica l'immagine nella variabile Airflow
image_name
.Utilizza DataprocCreateBatchOperator con la tua immagine personalizzata:
Utilizzo del servizio Dataproc Metastore con DataprocCreateBatchOperator
Per utilizzare un servizio Dataproc Metastore da un DAG:
Verifica che il servizio metastore sia già avviato.
Per scoprire come avviare un servizio metastore, consulta Abilitare e disabilitare Dataproc Metastore.
Per informazioni dettagliate sull'operatore batch per creare la configurazione, consulta PeripheralsConfig.
Quando il servizio metastore è attivo e in esecuzione, specifica il suo nome nella variabile
metastore_cluster
e la sua regione nella variabile Airflow diregion_name
.Usa il servizio metastore in DataprocCreateBatchOperator:
DataprocDeleteBatchOperator
Puoi utilizzare DataprocDeleteBatchOperator per eliminare un batch in base al relativo ID.
DataprocListBatchesOperator
DataprocDeleteBatchOperator elenca i batch esistenti all'interno di un progetto_id e di una regione specifici.
DataprocGetBatchOperator
DataprocGetBatchOperator recupera un determinato carico di lavoro batch.