Overview

Airflow 2.8, released on January 30, 2024, introduces the Object Storage abstraction and improves event listeners.

Main Features

Object Storage

The new ObjectStoragePath API unifies access to S3, GCS and Azure Blob files with a pathlib-compatible interface.

python 
from airflow.io.path import ObjectStoragePath

# Unified interface for S3, GCS, Azure
path = ObjectStoragePath('s3://bucket/data/file.csv')

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

# List files
for p in ObjectStoragePath('s3://bucket/data/').iterdir():
    print(p.name)

Event listeners

Listeners let you react to task lifecycle events (success, failure, retry) for monitoring and alerting.

python 
from airflow.listeners import hookimpl

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

@hookimpl
def on_task_instance_failed(previous_state, task_instance, session):
    print(f'Task failed: {task_instance.task_id}')
    # Send alert

Sources