Data Pipelines with Apache Airflow

Core Concepts

DAG (Directed Acyclic Graph): Your pipeline definition
Tasks: Units of work (operators)
Operators: Templates for task types
Dependencies: Order of task execution
Connections: Stored credentials for external systems
Variables: Configuration stored in Airflow
XComs: Passing data between tasks

Basic DAG

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.sql import SQLExecuteQueryOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['data-alerts@company.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'user_activity_pipeline',
    default_args=default_args,
    description='Process daily user activity',
    schedule='0 2 * * *',     # 2 AM daily
    start_date=datetime(2024, 1, 1),
    catchup=False,             # Don't backfill missed runs
    tags=['user-activity', 'daily'],
    max_active_runs=1,         # Only one run at a time
) as dag:

    extract = PythonOperator(
        task_id='extract_events',
        python_callable=extract_user_events,
        op_kwargs={'date': '{{ ds }}'},  # Jinja templating
    )

    transform = PythonOperator(
        task_id='transform_events',
        python_callable=transform_events,
    )

    load = PostgresOperator(
        task_id='load_to_warehouse',
        postgres_conn_id='postgres_warehouse',
        sql='sql/load_user_activity.sql',
        parameters={'date': '{{ ds }}'},
    )

    validate = PythonOperator(
        task_id='validate_results',
        python_callable=validate_row_counts,
    )

    # Define dependencies
    extract >> transform >> load >> validate

Dynamic DAGs

from airflow.models import Variable
import json

# Generate DAGs from config
configs = json.loads(Variable.get('pipeline_configs'))

for config in configs:
    with DAG(f"pipeline_{config['name']}", ...) as dag:
        tasks = []
        for step in config['steps']:
            task = PythonOperator(
                task_id=step['id'],
                python_callable=get_operator(step['type']),
                op_kwargs=step.get('params', {}),
            )
            tasks.append(task)

        # Chain tasks
        for i in range(len(tasks) - 1):
            tasks[i] >> tasks[i + 1]

XComs (Cross-Task Communication)

def extract_data(ti):  # ti = TaskInstance
    data = fetch_from_api()
    # Push to XCom
    ti.xcom_push(key='row_count', value=len(data))
    return data  # Return value is also pushed automatically

def validate_data(ti):
    row_count = ti.xcom_pull(task_ids='extract_data', key='row_count')
    if row_count < 1000:
        raise ValueError(f"Expected >1000 rows, got {row_count}")

# In Jinja templates:
# {{ ti.xcom_pull(task_ids='extract_data') }}

Error Handling and Alerting

from airflow.utils.email import send_email

def on_failure_callback(context):
    task = context['task_instance']
    dag = context['dag']
    exception = context['exception']

    send_email(
        to='data-team@company.com',
        subject=f"Airflow FAILURE: {dag.dag_id}.{task.task_id}",
        html_content=f"""
        <h2>Task Failed</h2>
        <p>DAG: {dag.dag_id}</p>
        <p>Task: {task.task_id}</p>
        <p>Run: {context['execution_date']}</p>
        <p>Error: {str(exception)}</p>
        """
    )

default_args = {
    'on_failure_callback': on_failure_callback,
    'on_retry_callback': on_retry_callback,
}

Kubernetes Executor

# docker-compose.yaml for local dev
services:
  airflow-webserver:
    image: apache/airflow:2.8.0
    command: webserver
    ports:
      - "8080:8080"
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow

  airflow-scheduler:
    image: apache/airflow:2.8.0
    command: scheduler
    depends_on:
      - airflow-webserver

Data Quality Checks

from great_expectations.dataset import PandasDataset

def validate_data_quality(ti):
    df = ti.xcom_pull(task_ids='transform_data')
    dataset = PandasDataset(df)

    # Expectations
    dataset.expect_column_values_to_not_be_null('user_id')
    dataset.expect_column_values_to_be_between('amount', 0, 100000)
    dataset.expect_column_values_to_be_unique('transaction_id')

    results = dataset.validate()
    if not results['success']:
        failed = [r for r in results['results'] if not r['success']]
        raise ValueError(f"Data quality failed: {failed}")

Airflow is the de facto standard for workflow orchestration - start simple, add complexity as needed.