Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
Questa guida mostra come scrivere un grafo diretto aciclico (DAG) di Apache Airflow in esecuzione in un ambiente Cloud Composer.
Poiché Apache Airflow non fornisce un isolamento rigoroso di DAG e attività, consigliamo di utilizzare ambienti di produzione e di test separati per evitare interferenze con il DAG. Per ulteriori informazioni, consulta Test dei DAG.
Strutturare un DAG Airflow
Un DAG Airflow è definito in un file Python ed è composto da quanto segue componenti:
- Definizione di DAG
- Operatori Airflow
- Relazioni con gli operatori
I seguenti snippet di codice mostrano esempi di ogni componente fuori contesto.
Una definizione di DAG
L'esempio seguente mostra un DAG Airflow definizione:
Flusso d'aria 2
Flusso d'aria 1
Operatori e attività
Gli operatori Airflow descrivono il lavoro da svolgere. Un'attività è un'istanza specifica di un operatore.
Airflow 2
Airflow 1
Relazioni tra attività
Le relazioni tra attività descrivono l'ordine in cui deve essere completato il lavoro.
Airflow 2
Flusso d'aria 1
Esempio di flusso di lavoro DAG completo in Python
Il seguente flusso di lavoro è un modello DAG completo e funzionante, composto da
due attività: un'attività hello_python
e un'attività goodbye_bash
:
Flusso d'aria 2
Airflow 1
Per saperne di più sulla definizione dei DAG Airflow, consulta Tutorial di Airflow e Concetti di Airflow.
Operatori Airflow
Gli esempi riportati di seguito mostrano alcuni operatori Airflow molto utilizzati. Per un riferimento autorevole degli operatori Airflow, consulta Riferimento per operatori e hook e l'indice dei provider.
BashOperator
Utilizza BashOperator per eseguire programmi a riga di comando.
Flusso d'aria 2
Flusso d'aria 1
Cloud Composer esegue i comandi forniti in uno script Bash su un Worker Airflow. Il worker è un container Docker basato su Debian e include diversi pacchetti.
- Comando
gcloud
, inclusogcloud storage
per lavorare con i bucket Cloud Storage. - Comando
bq
- Comando
kubectl
PythonOperator
Utilizza PythonOperator per eseguire codice Python arbitrario.
Cloud Composer esegue il codice Python in un container che include per la versione immagine di Cloud Composer utilizzati in del tuo ambiente.
Per installare altri pacchetti Python, consulta Installazione delle dipendenze Python.
Operatori Google Cloud
Per eseguire attività che utilizzano i prodotti Google Cloud, utilizza gli operatori Airflow di Google Cloud. Ad esempio: Operatori di BigQuery interrogare ed elaborare i dati in BigQuery.
Esistono molti altri operatori Airflow per Google Cloud e per i singoli servizi forniti da Google Cloud. Per l'elenco completo, consulta Operatori Google Cloud.
Airflow 2
Airflow 1
EmailOperator
Utilizza la EmailOperator per inviare email da un DAG. Per inviare le email da un ambiente Cloud Composer, configurare l'ambiente per l'utilizzo di SendGrid.
Airflow 2
Flusso d'aria 1
Notifiche in caso di errore dell'operatore
Imposta email_on_failure
su True
per inviare una notifica via email quando un operatore nel DAG non funziona. Per inviare notifiche email da Cloud Composer
dell'ambiente di lavoro,
configurare l'ambiente per l'utilizzo di SendGrid.
Airflow 2
Airflow 1
Linee guida per i flussi di lavoro DAG
Inserisci eventuali librerie Python personalizzate nell'archivio ZIP di un DAG in una directory nidificata. Non inserire le librerie al livello superiore della directory dei DAG.
Quando Airflow analizza la cartella
dags/
, verifica solo la presenza di DAG in I moduli Python che si trovano al primo livello della cartella dei DAG e in alto livello di un archivio ZIP situato anche nella cartella di primo livellodags/
. Se Airflow rileva un modulo Python in un archivio ZIP che non contiene entrambe le sottostringheairflow
eDAG
, interrompe l'elaborazione dell'archivio ZIP. Airflow restituisce solo i DAG trovati fino a quel punto.Utilizza Airflow 2 anziché Airflow 1.
La community di Airflow non pubblica più nuove release secondarie o patch per Airflow 1.
Per la tolleranza di errore, non definire più oggetti DAG nello stesso codice in maggior dettaglio più avanti in questo modulo.
Non utilizzare DAG secondari. Invece, raggruppa le attività all'interno dei DAG.
Inserisci i file richiesti al momento dell'analisi del DAG nella cartella
dags/
, non nella cartelladata/
.Testa i DAG sviluppati o modificati come consigliato nelle istruzioni per il test dei DAG.
Verifica che i DAG sviluppati non aumentino Tempi di analisi DAG eccessivi.
Le attività Airflow possono non riuscire per diversi motivi. Per evitare errori di esecuzione di interi DAG, ti consigliamo di attivare i tentativi di esecuzione delle attività. Se imposti il numero massimo di tentativi su
0
, non viene eseguito alcun tentativo.Ti consigliamo di sostituire il valore
default_task_retries
con un valore per il parametro con tentativi diversi da quelli di0
. Inoltre, puoi impostare Parametroretries
a livello di attività.Se vuoi utilizzare la GPU nelle attività di Airflow, crea un cluster GKE separato in base ai nodi che utilizzano macchine con GPU. Utilizza GKEStartPodOperator per eseguire le attività.
Evita di eseguire attività che richiedono molta CPU e memoria nel pool di nodi del cluster in cui sono in esecuzione altri componenti di Airflow (scheduler, worker, server web). Utilizza invece KubernetesPodOperator o GKEStartPodOperator.
Quando esegui il deployment dei DAG in un ambiente, carica nella cartella
/dags
solo i file assolutamente necessari per interpretare ed eseguire i DAG.Limita il numero di file DAG nella cartella
/dags
.Airflow analizza continuamente i DAG nella cartella
/dags
. L'analisi è un che esegue il loop attraverso la cartella dei DAG e il numero di file che che devono essere caricati (con le loro dipendenze) influisce sulle prestazioni dell'analisi dei DAG e della pianificazione delle attività. È molto più efficiente utilizzare 100 file con 100 DAG ciascuno rispetto a 10000 file con 1 DAG ciascuno, pertanto questa ottimizzazione è consigliata. Questa ottimizzazione è un equilibrio tra tempi di analisi ed efficienza della creazione e della gestione dei DAG.Puoi anche prendere in considerazione, ad esempio, di eseguire il deployment di 10000 file DAG creando 100 file ZIP contenenti ciascuno 100 file DAG.
Oltre ai suggerimenti precedenti, se hai più di 10.000 file DAG, generare i DAG in modo programmatico potrebbe essere una buona opzione. Ad esempio, puoi implementare un singolo file DAG Python che genera un certo numero di oggetti DAG (ad esempio 20, 100 oggetti DAG).
Evita di utilizzare operatori Airflow deprecati
Gli operatori elencati nella seguente tabella sono deprecati. Alcuni di questi operatori erano supportati nelle prime versioni di Cloud Composer 1. Evita di utilizzarli in dei tuoi DAG. Utilizza invece le alternative fornite e aggiornate.
Operatore ritirato | Operatore da utilizzare |
---|---|
BigQueryExecuteQueryOperator | BigQueryInsertJobOperator |
BigQueryPatchDatasetOperator | BigQueryUpdateTableOperator |
DataflowCreateJavaJobOperator | BeamRunJavaPipelineOperator |
DataflowCreatePythonJobOperator | BeamRunPythonPipelineOperator |
DataprocScaleClusterOperator | DataprocUpdateClusterOperator |
DataprocSubmitPigJobOperator | DataprocSubmitJobOperator |
DataprocSubmitSparkSqlJobOperator | DataprocSubmitJobOperator |
DataprocSubmitSparkJobOperator | DataprocSubmitJobOperator |
DataprocSubmitHadoopJobOperator | DataprocSubmitJobOperator |
DataprocSubmitPySparkJobOperator | DataprocSubmitJobOperator |
MLEngineManageModelOperator | MLEngineCreateModelOperator, MLEngineGetModelOperator |
MLEngineManageVersionOperator | MLEngineCreateVersion, MLEngineSetDefaultVersion, MLEngineListVersions, MLEngineDeleteVersion |
GCSObjectsWtihPrefixExistenceSensor | GCSObjectsWithPrefixExistenceSensor |
Domande frequenti sulla scrittura dei DAG
Come faccio a ridurre al minimo le ripetizioni del codice se voglio eseguire attività uguali o simili in più DAG?
Ti consigliamo di definire librerie e wrapper per al minimo le ripetizioni del codice.
Come posso riutilizzare il codice tra i file DAG?
Inserisci le funzioni di utilità in una
libreria Python locale
e importa le funzioni. Puoi fare riferimento alle funzioni in qualsiasi DAG situato
nella cartella dags/
del bucket del tuo ambiente.
Come faccio a ridurre al minimo il rischio che possano insorgere definizioni diverse?
Ad esempio, hai due team che vogliono aggregare i dati non elaborati in metriche sulle entrate. I team scrivono due attività leggermente diverse che svolgono la stessa cosa. Definisci le librerie lavorare con i dati sulle entrate in modo che gli implementatori DAG debbano chiarire la definizione di entrate aggregate.
Come posso impostare le dipendenze tra i DAG?
Questo dipende da come vuoi definire la dipendenza.
Se hai due DAG (DAG A e DAG B) e vuoi che il DAG B si attivi dopo il DAG
A, puoi inserire un
TriggerDagRunOperator
alla fine del DAG A.
Se il DAG B dipende solo da un artefatto generato dal DAG A, ad esempio un messaggio Pub/Sub, un sensore potrebbe funzionare meglio.
Se il DAG B è integrato strettamente con il DAG A, potresti essere in grado di unire i due i DAG in un DAG.
Come posso passare ID esecuzione univoci a un DAG e alle sue attività?
Ad esempio, vuoi passare i nomi dei cluster Dataproc e i percorsi dei file.
Puoi generare un ID univoco casuale restituendo str(uuid.uuid4())
in
un PythonOperator
. L'ID viene inserito in
XComs
in modo da poter fare riferimento all'ID in altri operatori
tramite i campi basati su modelli.
Prima di generare un uuid
, valuta se un ID specifico per DagRun sarebbe più utile. Puoi anche fare riferimento a questi ID nelle sostituzioni Jinja utilizzando le macro.
Come si separano le attività in un DAG?
Ogni attività deve essere un'unità di lavoro idempotente. Di conseguenza, dovresti evitare
incapsulare un flusso di lavoro in più fasi all'interno di una singola attività, ad esempio una
programma in esecuzione su PythonOperator
.
Devo definire più attività in un singolo DAG per aggregare dati da più origini?
Ad esempio, hai più tabelle con dati non elaborati e vuoi creare aggregati giornalieri per ogni tabella. Le attività non dipendono l'una dall'altra. Dovrei crei un'attività e un DAG per ogni tabella o crei un DAG generale?
Se ti va bene che ogni attività condivida le stesse proprietà a livello di DAG, come
schedule_interval
, allora ha senso definire più attività in un singolo
con il DAG. In caso contrario, per ridurre al minimo la ripetizione del codice, è possibile generare più DAG
da un singolo modulo Python inserendoli nel globals()
del modulo.
Come posso limitare il numero di attività simultanee in esecuzione in un DAG?
Ad esempio, vuoi evitare di superare i limiti/quote di utilizzo delle API eseguendo troppi processi simultanei.
Puoi definire pool Airflow nell'interfaccia utente web di Airflow e associare le attività ai pool esistenti nei tuoi DAG.
Domande frequenti sull'utilizzo degli operatori
Dovrei usare DockerOperator
?
Ti sconsigliamo di utilizzare
DockerOperator
, a meno che non venga utilizzato per avviare
contenuti su un'installazione Docker remota (non all'interno del
cluster di un ambiente). In un ambiente Cloud Composer l'operatore non ha
ai daemon Docker.
Usa invece KubernetesPodOperator
o
GKEStartPodOperator
. Questi operatori lanciano i pod Kubernetes rispettivamente nei cluster Kubernetes o GKE. Tieni presente che non consigliamo di avviare pod nel cluster di un ambiente, perché ciò può portare alla concorrenza per le risorse.
Devo utilizzare SubDagOperator
?
Ti consigliamo di non utilizzare SubDagOperator
.
Utilizza le alternative suggerite in Raggruppare le attività.
Devo eseguire il codice Python solo in PythonOperators
per separare completamente gli operatori Python?
A seconda del tuo obiettivo, hai a disposizione alcune opzioni.
Se la tua unica preoccupazione è mantenere dipendenze Python separate,
puoi usare PythonVirtualenvOperator
.
Valuta la possibilità di utilizzare KubernetesPodOperator
. Questo operatore ti consente di definire i pod Kubernetes ed eseguirli in altri cluster.
Come faccio ad aggiungere pacchetti non PyPI o binari personalizzati?
Puoi installare pacchetti ospitati in repository di pacchetti privati.
Come faccio a passare gli argomenti in modo uniforme a un DAG e alle relative attività?
Puoi utilizzare il supporto integrato di Airflow per il modello Jinja per passare argomenti che possono essere utilizzati nei campi basati su modelli.
Quando avviene la sostituzione del modello?
La sostituzione del modello avviene sui worker di Airflow appena prima della chiamata della funzione pre_execute
di un operatore. In pratica, ciò significa che i modelli
non viene sostituito fino a poco prima dell'esecuzione di un'attività.
Come faccio a sapere quali argomenti dell'operatore supportano la sostituzione del modello?
Gli argomenti dell'operatore che supportano la sostituzione del modello Jinja2 sono esplicitamente contrassegnati come tali.
Cerca il campo template_fields
nella definizione dell'operatore,
che contiene un elenco di nomi di argomenti sottoposti alla sostituzione del modello.
Ad esempio, consulta
BashOperator
, che supporta la creazione di modelli per
gli argomenti bash_command
e env
.
Passaggi successivi
- Risoluzione dei problemi dei DAG
- Risoluzione dei problemi relativi alla pianificazione
- Operatori Google
- Operatori Google Cloud
- Tutorial su Apache Airflow