Tag: airflow, python, dag, etl, orchestration, pipeline

Livello: intermedio

TL;DR: Riferimento rapido su Apache Airflow: componenti principali, concetti di DAG e operatori, gestione dello stato e best practice per i task.

Concetti chiave

-

Airflow è una piattaforma open-source per orchestrare, schedulare e monitorare workflow programmaticamente.

-

Un DAG (Directed Acyclic Graph) definisce il workflow: una collezione di task con dipendenze direzionali e nessun ciclo.

-

Gli Operator determinano cosa viene effettivamente eseguito (es. PythonOperator, BashOperator).

-

I Sensor sono operatori speciali progettati per rimanere in attesa di una condizione o evento esterno.

-

Gli Hook offrono un'interfaccia standardizzata per connettersi a sistemi e servizi esterni (database, cloud provider).

-

Il Webserver espone la UI per monitorare, attivare e fare il debug di DAG e task.

-

Lo Scheduler analizza i DAG e invia in esecuzione i task le cui dipendenze sono state soddisfatte.

-

Gli XCom (Cross-Communication) permettono lo scambio di piccoli payload o metadati tra i task.

-

L'Executor (es. LocalExecutor, CeleryExecutor, KubernetesExecutor) decide dove e come i task allocati vengono eseguiti fisicamente.

Domande tipiche e risposte pronte

-

Q: Come scambi i dati tra due task in Airflow?R: Uso gli XCom per passare stringhe, ID o piccoli metadati. Per i dataset veri e propri salvo su un object storage (S3/GCS) e passo solo l'URI tramite XCom.

-

Q: Che differenza c'è tra Operator e Task?R: L'Operator è la classe o il template che definisce il lavoro da svolgere. Un Task è un'istanza concreta di quell'Operator posizionata all'interno di un DAG.

-

Q: Come gestisci un task che fallisce a intermittenza a causa di limiti sulle API esterne?R: Sfrutto i parametri retries e retry_delay nel costruttore dell'operatore, per garantire che Airflow riprovi l'esecuzione dopo un'attesa, assorbendo il problema temporaneo.

Insidie e errori comuni

-

Inserire codice pesante, query a DB o chiamate di rete nel livello top-level del file Python: rallenta e blocca il parsing continuo dello scheduler.

-

Utilizzare gli XCom per passare DataFrame interi o blob di dati, sovraccaricando e intasando il database dei metadati di Airflow.

-

Progettare task che non sono idempotenti: ogni riesecuzione del task in caso di fallimento deve produrre esattamente lo stesso risultato senza duplicazioni.

-

Dimenticare di configurare il parametro execution_timeout per task soggetti a blocchi (es. API esterne silenziose), bloccando di fatto gli slot dei worker.

Snippet e comandiPython

```from airflow import DAG

from airflow.operators.python import PythonOperator

from datetime import datetime, timedelta

default_args = {

'owner': 'airflow',

'retries': 2,

'retry_delay': timedelta(minutes=5),

}

with DAG(

'example_data_pipeline',

default_args=default_args,

description='A basic ETL DAG',

schedule_interval='0 2 * * *', # Ogni giorno alle 02:00

start_date=datetime(2023, 1, 1),

catchup=False,

tags=['example', 'etl']

) as dag:

def extract_data(**kwargs):

print("Extracting data...")

return "data_uri_123"

extract_task = PythonOperator(

task_id='extract',

python_callable=extract_data,

)

Testare a vuoto l'esecuzione di un singolo task localmente (senza registrarlo a DB)

airflow tasks test example_data_pipeline extract 2023-01-01

Sospendere o riattivare un DAG da CLI

airflow dags pause example_data_pipeline

airflow dags unpause example_data_pipeline

Elencare i DAG attivi

airflow dags list

```