Vue d'ensemble

Airflow 2.8, publie le 30 janvier 2024, introduit l'abstraction Object Storage et ameliore les listeners d'evenements.

Fonctionnalites principales

Object Storage

La nouvelle API ObjectStoragePath unifie l'acces aux fichiers S3, GCS et Azure Blob avec une interface compatible pathlib.

python 
from airflow.io.path import ObjectStoragePath

# Interface unifiee pour S3, GCS, Azure
path = ObjectStoragePath('s3://bucket/data/file.csv')

with path.open('r') as f:
    content = f.read()

# Lister les fichiers
for p in ObjectStoragePath('s3://bucket/data/').iterdir():
    print(p.name)

Listeners d'evenements

Les listeners permettent de reagir aux evenements du cycle de vie des taches (succes, echec, retry) pour le monitoring et l'alerting.

python 
from airflow.listeners import hookimpl

@hookimpl
def on_task_instance_success(previous_state, task_instance, session):
    print(f'Tache reussie: {task_instance.task_id}')

@hookimpl
def on_task_instance_failed(previous_state, task_instance, session):
    print(f'Tache echouee: {task_instance.task_id}')
    # Envoyer une alerte

Sources