Building a Data Pipeline with Python and Apache Airflow

Why Airflow

Cron jobs work fine until you need retries, dependency management between tasks, and visibility into what failed and why. Airflow solves all three by modeling pipelines as directed acyclic graphs (DAGs) of tasks.

A Complete ETL DAG

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

def extract(**context):
    import requests
    response = requests.get('https://api.example.com/sales')
    context['ti'].xcom_push(key='raw_data', value=response.json())

def transform(**context):
    raw = context['ti'].xcom_pull(key='raw_data', task_ids='extract')
    cleaned = [row for row in raw if row.get('amount', 0) > 0]
    context['ti'].xcom_push(key='clean_data', value=cleaned)

def load(**context):
    clean = context['ti'].xcom_pull(key='clean_data', task_ids='transform')
    import psycopg2
    conn = psycopg2.connect("dbname=warehouse")
    cur = conn.cursor()
    for row in clean:
        cur.execute(
            "INSERT INTO sales (id, amount) VALUES (%s, %s)",
            (row['id'], row['amount'])
        )
    conn.commit()

with DAG(
    'daily_sales_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:
    extract_task = PythonOperator(task_id='extract', python_callable=extract)
    transform_task = PythonOperator(task_id='transform', python_callable=transform)
    load_task = PythonOperator(task_id='load', python_callable=load)

    extract_task >> transform_task >> load_task

Key Concepts

• XCom — passes small amounts of data between tasks; not meant for large payloads.
• Retries — configured per-task, so transient API failures don’t fail the entire pipeline.
• catchup=False — prevents Airflow from backfilling every missed run when you first deploy a DAG.

Handling Large Data Volumes

For large datasets, avoid pushing raw data through XCom — write intermediate results to cloud storage (S3, GCS) and pass file references between tasks instead.

Monitoring and Alerting

default_args = {
    'on_failure_callback': send_slack_alert,
}

Wire failure callbacks to Slack or PagerDuty so pipeline failures don’t go unnoticed until someone checks a dashboard.

Conclusion

Airflow’s DAG model makes pipeline dependencies and failure handling explicit rather than implicit in a tangle of cron jobs. Start with a simple linear pipeline, and only reach for more advanced patterns (dynamic task mapping, sensors) once you have a concrete need for them.