Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1
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 forte isolamento di DAG e attività, ti consigliamo di utilizzare ambienti di produzione e test separati per evitare interferenze tra i DAG. Per ulteriori informazioni, consulta Testare i DAG.
Strutturare un DAG Airflow
Un DAG di Airflow è definito in un file Python ed è composto dai seguenti componenti:
- Definizione del DAG
- Operatori Airflow
- Rapporti con gli operatori
I seguenti snippet di codice mostrano esempi di ciascun componente fuori contesto.
Definizione di un DAG
L'esempio seguente mostra una definizione di DAG di Airflow:
Operatori e attività
Gli operatori Airflow descrivono il lavoro da svolgere. Un'attività task è un'istanza specifica di un operatore.
Relazioni tra attività
Le relazioni tra le attività descrivono l'ordine in cui il lavoro deve essere completato.
Esempio di flusso di lavoro DAG completo in Python
Il seguente workflow è un modello DAG funzionante completo composto da due attività: un'attività hello_python
e un'attività goodbye_bash
:
Per ulteriori informazioni sulla definizione dei DAG di Airflow, consulta il tutorial su Airflow e i concetti di Airflow.
Operatori Airflow
Gli esempi seguenti mostrano alcuni operatori Airflow popolari. Per un riferimento autorevole degli operatori Airflow, consulta la Guida di riferimento di operatori e hook e l'indice dei provider.
BashOperator
Utilizza BashOperator per eseguire programmi a riga di comando.
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
, incluso il sottocomandogcloud 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 i pacchetti per la versione dell'immagine Cloud Composer utilizzata nel tuo ambiente.
Per installare pacchetti Python aggiuntivi, consulta Installazione delle dipendenze Python.
Google Cloud Operatori
Per eseguire attività che utilizzano i prodotti Google Cloud , utilizza gli operatori AirflowGoogle Cloud . Ad esempio, gli operatori BigQuery eseguono query ed elaborano i dati in BigQuery.
Esistono molti altri operatori Airflow per Google Cloud e singoli servizi forniti da Google Cloud. Per l'elenco completo, vedi Google Cloud Operatori.
EmailOperator
Utilizza EmailOperator per inviare email da un DAG. Per inviare email da un ambiente Cloud Composer, configura l'ambiente per utilizzare SendGrid.
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 riesce. Per inviare notifiche via email da un ambiente Cloud Composer, devi configurare l'ambiente in modo che utilizzi SendGrid.
Linee guida per il flusso di lavoro DAG
Inserisci le librerie Python personalizzate nell'archivio ZIP di un DAG in una directory nidificata. Non inserire le librerie nel livello superiore della directory DAG.
Quando Airflow esegue la scansione della cartella
dags/
, controlla solo i DAG nei moduli Python che si trovano nel livello superiore della cartella DAG e nel livello superiore di un archivio ZIP che si trova anche nella cartelladags/
di primo livello. Se Airflow rileva un modulo Python in un archivio ZIP che non contiene le sottostringheairflow
eDAG
, interrompe l'elaborazione dell'archivio ZIP. Airflow restituisce solo i DAG trovati fino a quel momento.Per la tolleranza agli errori, non definire più oggetti DAG nello stesso modulo Python.
Non utilizzare i SubDAG. In alternativa, 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 troppo i tempi di analisi dei DAG.
Le attività Airflow possono non riuscire per diversi motivi. Per evitare errori durante l'esecuzione dell'intero DAG, ti consigliamo di attivare i tentativi di ripetizione delle attività. Se imposti il numero massimo di tentativi su
0
, non vengono eseguiti tentativi.Ti consigliamo di eseguire l'override dell'opzione
default_task_retries
con un valore per i tentativi di ripetizione dell'attività diverso da0
. Inoltre, puoi impostare il parametroretries
a livello di attività.Se vuoi utilizzare la GPU nelle attività Airflow, crea un cluster GKE separato basato su nodi che utilizzano macchine con GPU. Utilizza GKEStartPodOperator per eseguire le attività.
Evita di eseguire attività che richiedono un utilizzo elevato di 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 solo i file assolutamente necessari per interpretare ed eseguire i DAG nella cartella
/dags
.Limita il numero di file DAG nella cartella
/dags
.Airflow analizza continuamente i DAG nella cartella
/dags
. L'analisi è un processo che scorre la cartella DAG e il numero di file da caricare (con le relative dipendenze) influisce sulle prestazioni dell'analisi DAG e della pianificazione delle attività. È molto più efficiente utilizzare 100 file con 100 DAG ciascuno rispetto a 10.000 file con 1 DAG ciascuno, pertanto è consigliata questa ottimizzazione. Questa ottimizzazione è un equilibrio tra tempo di analisi ed efficienza della creazione e della gestione dei DAG.Ad esempio, se vuoi eseguire il deployment di 10.000 file DAG, puoi creare 100 file ZIP contenenti 100 file DAG ciascuno.
Oltre ai suggerimenti riportati sopra, se hai più di 10.000 file DAG, la generazione di 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 ritirati. In alternativa, utilizza le alternative aggiornate.
Domande frequenti sulla scrittura dei DAG
Come faccio a ridurre al minimo la ripetizione del codice se voglio eseguire attività uguali o simili in più DAG?
Ti consigliamo di definire librerie e wrapper per ridurre al minimo la ripetizione del codice.
Come faccio a 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
che si trova nella cartella dags/
del bucket del tuo ambiente.
Come faccio a ridurre al minimo il rischio di definizioni diverse?
Ad esempio, hai due team che vogliono aggregare i dati non elaborati in metriche relative alle entrate. I team scrivono due attività leggermente diverse che ottengono lo stesso risultato. Definisci le librerie per lavorare con i dati sulle entrate in modo che gli implementatori del DAG debbano chiarire la definizione delle entrate aggregate.
Come faccio a impostare le dipendenze tra i DAG?
Dipende da come vuoi definire la dipendenza.
Se hai due DAG (DAG A e DAG B) e vuoi che il DAG B venga attivato 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 è strettamente integrato con il DAG A, potresti essere in grado di unire i due DAG in un unico DAG.
Come faccio a trasmettere ID esecuzione univoci a un DAG e ai relativi task?
Ad esempio, vuoi trasmettere i nomi dei cluster Dataproc e i percorsi dei file.
Puoi generare un ID univoco casuale restituendo str(uuid.uuid4())
in
un PythonOperator
. In questo modo l'ID viene inserito in
XComs
, in modo che tu possa farvi riferimento 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 faccio a separare le attività in un DAG?
Ogni attività deve essere un'unità di lavoro idempotente. Di conseguenza, devi evitare
di incapsulare un flusso di lavoro in più passaggi in un'unica attività, ad esempio un programma
complesso in esecuzione in un PythonOperator
.
Devo definire più attività in un singolo DAG per aggregare i dati di più origini?
Ad esempio, hai più tabelle con dati non elaborati e vuoi creare aggregazioni giornaliere per ogni tabella. Le attività non dipendono l'una dall'altra. Devi creare un'attività e un DAG per ogni tabella o un DAG generale?
Se non hai problemi a condividere le stesse proprietà a livello di DAG per ogni attività, ad esempio
schedule_interval
, allora è opportuno definire più attività in un singolo
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 faccio a limitare il numero di attività simultanee in esecuzione in un DAG?
Ad esempio, vuoi evitare di superare i limiti/le quote di utilizzo dell'API o di eseguire 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
Devo utilizzare DockerOperator
?
Ti sconsigliamo di utilizzare
DockerOperator
, a meno che non venga utilizzato per avviare
i container su un'installazione Docker remota (non all'interno del cluster di un ambiente). In un ambiente Cloud Composer l'operatore non ha accesso ai daemon Docker.
Utilizza invece KubernetesPodOperator
o
GKEStartPodOperator
. Questi operatori avviano i pod Kubernetes nei cluster Kubernetes o GKE, rispettivamente. Tieni presente che non
consigliamo di avviare i pod nel cluster di un ambiente, perché ciò può portare
a una competizione per le risorse.
Devo utilizzare SubDagOperator
?
Non è consigliabile utilizzare SubDagOperator
.
Utilizza le alternative suggerite in Raggruppamento delle attività.
Devo eseguire il codice Python solo in PythonOperators
per separare completamente gli operatori Python?
A seconda dell'obiettivo, hai a disposizione alcune opzioni.
Se la tua unica preoccupazione è mantenere dipendenze Python separate, puoi utilizzare 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 binari personalizzati o non PyPI?
Puoi installare pacchetti ospitati in Package Repository privati.
Come faccio a passare argomenti in modo uniforme a un DAG e alle relative attività?
Puoi utilizzare il supporto integrato di Airflow per i modelli Jinja per passare argomenti che possono essere utilizzati nei campi basati su modelli.
Quando avviene la sostituzione del modello?
La sostituzione dei modelli avviene sui worker Airflow appena prima della chiamata alla funzione pre_execute
di un operatore. In pratica, i modelli non vengono sostituiti fino a poco prima dell'esecuzione di un'attività.
Come faccio a sapere quali argomenti dell'operatore supportano la sostituzione dei modelli?
Gli argomenti dell'operatore che supportano la sostituzione del modello Jinja2 sono contrassegnati esplicitamente come tali.
Cerca il campo template_fields
nella definizione dell'operatore,
che contiene un elenco di nomi di argomenti che vengono sostituiti dal modello.
Ad esempio, consulta
BashOperator
, che supporta i modelli per
gli argomenti bash_command
e env
.
Operatori Airflow deprecati e rimossi
Gli operatori Airflow elencati nella tabella seguente sono ritirati:
Evita di utilizzare questi operatori nei tuoi DAG. Utilizza invece gli operatori di sostituzione aggiornati forniti.
Se un operatore è elencato come rimosso, significa che non è più disponibile in una delle build di Airflow rilasciate in Cloud Composer 3.
Se un operatore è elencato come pianificato per la rimozione, è deprecato e verrà rimosso in una build di Airflow futura.
Se un operatore è elencato come già rimosso negli ultimi fornitori Google, allora l'operatore viene rimosso nell'ultima versione del pacchetto
apache-airflow-providers-google
. Allo stesso tempo, Cloud Composer utilizza ancora la versione di questo pacchetto in cui l'operatore non è ancora stato rimosso.
Operatore ritirato | Stato | Operatore sostitutivo | Sostituzione disponibile a partire dal giorno |
---|---|---|---|
CreateAutoMLTextTrainingJobOperator | Rimosso | SupervisedFineTuningTrainOperator |
composer-3-airflow-2.9.3-build.1 composer-3-airflow-2.9.1-build.8 |
GKEDeploymentHook | Rimosso | GKEKubernetesHook |
Tutte le versioni |
GKECustomResourceHook | Rimosso | GKEKubernetesHook |
Tutte le versioni |
GKEPodHook | Rimosso | GKEKubernetesHook |
Tutte le versioni |
GKEJobHook | Rimosso | GKEKubernetesHook |
Tutte le versioni |
GKEPodAsyncHook | Rimosso | GKEKubernetesAsyncHook |
Tutte le versioni |
SecretsManagerHook | Rimosso | GoogleCloudSecretManagerHook |
composer-3-airflow-2.7.3-build.6 |
BigQueryExecuteQueryOperator | Rimosso | BigQueryInsertJobOperator |
Tutte le versioni |
BigQueryPatchDatasetOperator | Rimosso | BigQueryUpdateDatasetOperator |
Tutte le versioni |
DataflowCreateJavaJobOperator | Rimosso | beam.BeamRunJavaPipelineOperator |
Tutte le versioni |
DataflowCreatePythonJobOperator | Rimosso | beam.BeamRunPythonPipelineOperator |
Tutte le versioni |
DataprocSubmitPigJobOperator | Rimosso | DataprocSubmitJobOperator |
Tutte le versioni |
DataprocSubmitHiveJobOperator | Rimosso | DataprocSubmitJobOperator |
Tutte le versioni |
DataprocSubmitSparkSqlJobOperator | Rimosso | DataprocSubmitJobOperator |
Tutte le versioni |
DataprocSubmitSparkJobOperator | Rimosso | DataprocSubmitJobOperator |
Tutte le versioni |
DataprocSubmitHadoopJobOperator | Rimosso | DataprocSubmitJobOperator |
Tutte le versioni |
DataprocSubmitPySparkJobOperator | Rimosso | DataprocSubmitJobOperator |
Tutte le versioni |
BigQueryTableExistenceAsyncSensor | Rimosso | BigQueryTableExistenceSensor |
Tutte le versioni |
BigQueryTableExistencePartitionAsyncSensor | Rimosso | BigQueryTablePartitionExistenceSensor |
Tutte le versioni |
CloudComposerEnvironmentSensor | Rimosso | CloudComposerCreateEnvironmentOperator, CloudComposerDeleteEnvironmentOperator, CloudComposerUpdateEnvironmentOperator |
Tutte le versioni |
GCSObjectExistenceAsyncSensor | Rimosso | GCSObjectExistenceSensor |
Tutte le versioni |
GoogleAnalyticsHook | Rimosso | GoogleAnalyticsAdminHook |
Tutte le versioni |
GoogleAnalyticsListAccountsOperator | Rimosso | GoogleAnalyticsAdminListAccountsOperator |
Tutte le versioni |
GoogleAnalyticsGetAdsLinkOperator | Rimosso | GoogleAnalyticsAdminGetGoogleAdsLinkOperator |
Tutte le versioni |
GoogleAnalyticsRetrieveAdsLinksListOperator | Rimosso | GoogleAnalyticsAdminListGoogleAdsLinksOperator |
Tutte le versioni |
GoogleAnalyticsDataImportUploadOperator | Rimosso | GoogleAnalyticsAdminCreateDataStreamOperator |
Tutte le versioni |
GoogleAnalyticsDeletePreviousDataUploadsOperator | Rimosso | GoogleAnalyticsAdminDeleteDataStreamOperator |
Tutte le versioni |
DataPipelineHook | Rimosso | DataflowHook |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
CreateDataPipelineOperator | Rimosso | DataflowCreatePipelineOperator |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
RunDataPipelineOperator | Rimosso | DataflowRunPipelineOperator |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
AutoMLDatasetLink | Obsoleto, rimozione pianificata | TranslationLegacyDatasetLink |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
AutoMLDatasetListLink | Obsoleto, rimozione pianificata | TranslationDatasetListLink |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
AutoMLModelLink | Obsoleto, rimozione pianificata | TranslationLegacyModelLink |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
AutoMLModelTrainLink | Obsoleto, rimozione pianificata | TranslationLegacyModelTrainLink |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
AutoMLModelPredictLink | Obsoleto, rimozione pianificata | TranslationLegacyModelPredictLink |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
AutoMLBatchPredictOperator | Rimosso | vertex_ai.batch_prediction_job |
composer-3-airflow-2.9.3-build.4 |
AutoMLPredictOperator | Obsoleto, rimozione pianificata | vertex_aigenerative_model. TextGenerationModelPredictOperator, translate.TranslateTextOperator |
composer-3-airflow-2.7.3-build.6 |
PromptLanguageModelOperator | Rimosso | TextGenerationModelPredictOperator |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
GenerateTextEmbeddingsOperator | Rimosso | TextEmbeddingModelGetEmbeddingsOperator |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
PromptMultimodalModelOperator | Rimosso | GenerativeModelGenerateContentOperator |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
PromptMultimodalModelWithMediaOperator | Rimosso | GenerativeModelGenerateContentOperator |
composer-3-airflow-2.9.1-build.0 composer-3-airflow-2.7.3-build.9 |
DataflowStartSqlJobOperator | Rimosso | DataflowStartYamlJobOperator |
composer-3-airflow-2.9.3-build.1 composer-3-airflow-2.9.1-build.8 |
LifeSciencesHook | Obsoleto, rimozione pianificata | Hook degli operatori batch di Google Cloud |
Da definire |
DataprocScaleClusterOperator | Obsoleto, rimozione pianificata | DataprocUpdateClusterOperator |
Da definire |
MLEngineStartBatchPredictionJobOperator | Obsoleto, rimozione pianificata | CreateBatchPredictionJobOperator |
Da definire |
MLEngineManageModelOperator | Obsoleto, rimozione pianificata | MLEngineCreateModelOperator, MLEngineGetModelOperator |
Da definire |
MLEngineGetModelOperator | Obsoleto, rimozione pianificata | GetModelOperator |
Da definire |
MLEngineDeleteModelOperator | Obsoleto, rimozione pianificata | DeleteModelOperator |
Da definire |
MLEngineManageVersionOperator | Obsoleto, rimozione pianificata | MLEngineCreateVersion, MLEngineSetDefaultVersion, MLEngineListVersions, MLEngineDeleteVersion |
Da definire |
MLEngineCreateVersionOperator | Obsoleto, rimozione pianificata | Parametro parent_model per gli operatori VertexAI |
Da definire |
MLEngineSetDefaultVersionOperator | Obsoleto, rimozione pianificata | SetDefaultVersionOnModelOperator |
Da definire |
MLEngineListVersionsOperator | Obsoleto, rimozione pianificata | ListModelVersionsOperator |
Da definire |
MLEngineDeleteVersionOperator | Obsoleto, rimozione pianificata | DeleteModelVersionOperator |
Da definire |
MLEngineStartTrainingJobOperator | Obsoleto, rimozione pianificata | CreateCustomPythonPackageTrainingJobOperator |
Da definire |
MLEngineTrainingCancelJobOperator | Obsoleto, rimozione pianificata | CancelCustomTrainingJobOperator |
Da definire |
LifeSciencesRunPipelineOperator | Obsoleto, rimozione pianificata | Operatori batch di Google Cloud |
Da definire |
MLEngineCreateModelOperator | Obsoleto, rimozione pianificata | operatore VertexAI corrispondente |
Da definire |
Passaggi successivi
- Risoluzione dei problemi dei DAG
- Risoluzione dei problemi dello strumento di pianificazione
- Operatori Google
- Google Cloud Operatori
- Tutorial su Apache Airflow