Apache Airflow Guide

# Set Airflow home
export AIRFLOW_HOME=~/airflow

AIRFLOW_VERSION=3.2.2
PYTHON_VERSION="$(python --version | cut -d ' ' -f 2 | cut -d '.' -f 1-2)"
CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"

pip install "apache-airflow==${AIRFLOW_VERSION}" \
    --constraint "${CONSTRAINT_URL}"

# Initialise the database
airflow db migrate

# Create admin user
airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email admin@example.com

# Start services in separate terminals
airflow webserver --port 8080
airflow scheduler

from datetime import datetime, timedelta
from airflow.sdk import DAG  # New unified import in 3.x
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'data-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='hello_airflow',
    default_args=default_args,
    schedule=timedelta(days=1),  # schedule_interval deprecated in 3.x
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['tutorial'],
) as dag:

    print_date = BashOperator(
        task_id='print_date',
        bash_command='date',
    )

    def greet():
        print("Hello from Airflow 3.x!")

    greet_task = PythonOperator(
        task_id='greet_user',
        python_callable=greet,
    )

    print_date >> greet_task

from datetime import datetime
from airflow.sdk import dag, task

@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False)
def my_pipeline():

    @task
    def extract():
        return {"records": 100, "status": "ok"}

    @task
    def transform(data: dict):
        data["processed"] = True
        return data

    @task
    def load(data: dict):
        print(f"Loading {data['records']} records")

    load(transform(extract()))

my_pipeline()

# Linear: A then B then C
task_a >> task_b >> task_c

# Fan-out: A triggers B, C, D simultaneously
task_a >> [task_b, task_c, task_d]

# Fan-in: B, C, D must complete before E
[task_b, task_c, task_d] >> task_e

# Complex chains
start >> [parallel_1, parallel_2] >> combine >> end

from datetime import datetime, timedelta
from airflow.sdk import dag, task
import pandas as pd, requests, json

@dag(
    schedule='0 3 * * *',
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={
        'owner': 'data-engineering',
        'retries': 3,
        'retry_delay': timedelta(minutes=5),
    },
)
def etl_pipeline():

    @task
    def extract(**context):
        url = f'https://api.example.com/data?date={context["ds"]}'
        data = requests.get(url, timeout=30).json()
        path = f'/tmp/raw_{context["ds"]}.json'
        with open(path, 'w') as f:
            json.dump(data, f)
        return path

    @task
    def transform(file_path: str, **context):
        df = pd.read_json(file_path)
        df['processed_date'] = pd.to_datetime(context['ds'])
        df['value_doubled'] = df['value'] * 2
        out = f'/tmp/transformed_{context["ds"]}.csv'
        df.to_csv(out, index=False)
        return out

    @task
    def load(file_path: str):
        import sqlite3
        df = pd.read_csv(file_path)
        conn = sqlite3.connect('/tmp/warehouse.db')
        df.to_sql('analytics', conn, if_exists='append', index=False)
        conn.close()

    load(transform(extract()))

etl_pipeline()

@task
def process_partition(partition_id: str):
    print(f"Processing {partition_id}")

# Tasks created dynamically at runtime
process_partition.expand(
    partition_id=['north', 'south', 'east', 'west']
)

from airflow.sdk import Asset

orders_asset = Asset("s3://my-bucket/orders/")

# Producer — publishes the asset
@dag(schedule="@daily")
def produce_orders():
    @task(outlets=[orders_asset])
    def write_orders():
        pass
    write_orders()

# Consumer — triggers when asset updates
@dag(schedule=[orders_asset])
def consume_orders():
    @task
    def process():
        print("New orders — processing now")
    process()

@task
def idempotent_load(**context):
    import sqlite3
    date = context['ds']
    conn = sqlite3.connect('/tmp/db.db')
    conn.execute(
        "DELETE FROM metrics WHERE date = ?", (date,)
    )
    df = pd.read_csv(f'/tmp/data_{date}.csv')
    df.to_sql('metrics', conn, if_exists='append', index=False)
    conn.commit()
    conn.close()

from airflow.models import Variable

# Store via CLI: airflow variables set api_key "value"
api_key = Variable.get("api_key")

# Configure DB connections via Admin > Connections in the UI
# Never hardcode credentials directly in DAG files

airflow dags list
airflow dags trigger etl_pipeline
airflow tasks test etl_pipeline extract 2026-01-15
airflow dags pause etl_pipeline
airflow dags unpause etl_pipeline
airflow dags backfill etl_pipeline -s 2026-01-01 -e 2026-01-31