Apache Airflow Guide — DAGs, Operators, and ETL Orchestration

# simple_dag.py
# Minimal Airflow DAG definition
from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator

default_args = {
    'owner': 'dodatech',
    'start_date': datetime(2026, 1, 1),
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'simple_etl_dag',
    default_args=default_args,
    description='Simple ETL pipeline',
    schedule_interval='@daily',
    catchup=False,
    tags=['etl', 'tutorial'],
)

# Defining tasks and dependencies
task_extract = PythonOperator(
    task_id='extract_data',
    python_callable=extract_function,
    dag=dag,
)

task_transform = PythonOperator(
    task_id='transform_data',
    python_callable=transform_function,
    dag=dag,
)

task_load = BashOperator(
    task_id='load_to_warehouse',
    bash_command='dbt run --models sales',
    dag=dag,
)

# Define execution order
task_extract >> task_transform >> task_load

# etl_dag.py
# Complete ETL DAG with PythonOperator and BashOperator
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator

default_args = {
    'owner': 'dodatech',
    'depends_on_past': False,
    'start_date': datetime(2026, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=10),
}

dag = DAG(
    'sales_etl_pipeline',
    default_args=default_args,
    description='Daily sales ETL: extract, transform, load',
    schedule_interval='0 2 * * *',  # Run daily at 2 AM
    catchup=False,
    tags=['sales', 'etl'],
)

def extract_sales(**context):
    """Extract sales data from API and write to staging."""
    import json
    import requests

    # Simulate API call
    api_response = {
        "orders": [
            {"order_id": 1001, "amount": 250.00, "status": "completed"},
            {"order_id": 1002, "amount": 45.50, "status": "pending"},
            {"order_id": 1003, "amount": 1200.00, "status": "completed"},
        ]
    }

    # Push to XCom for downstream tasks
    context['task_instance'].xcom_push(key='sales_data', value=api_response)
    print(f"Extracted {len(api_response['orders'])} orders")

def validate_sales(**context):
    """Validate extracted data before transformation."""
    ti = context['task_instance']
    data = ti.xcom_pull(task_ids='extract_sales', key='sales_data')
    errors = []

    for order in data['orders']:
        if order['amount'] <= 0:
            errors.append(f"Order {order['order_id']}: invalid amount")
        if order['status'] not in ('completed', 'pending', 'cancelled'):
            errors.append(f"Order {order['order_id']}: unknown status")

    if errors:
        raise ValueError(f"Validation errors: {errors}")
    print(f"Validation passed: {len(data['orders'])} orders OK")

def transform_sales(**context):
    """Clean and enrich sales data."""
    import json
    ti = context['task_instance']
    data = ti.xcom_pull(task_ids='extract_sales', key='sales_data')

    transformed = []
    for order in data['orders']:
        order['category'] = 'high_value' if order['amount'] > 100 else 'standard'
        order['processed_at'] = datetime.now().isoformat()
        transformed.append(order)

    ti.xcom_push(key='transformed_data', value=transformed)
    print(f"Transformed {len(transformed)} orders")

# Task definitions
extract_task = PythonOperator(
    task_id='extract_sales',
    python_callable=extract_sales,
    provide_context=True,
    dag=dag,
)

validate_task = PythonOperator(
    task_id='validate_sales',
    python_callable=validate_sales,
    provide_context=True,
    dag=dag,
)

transform_task = PythonOperator(
    task_id='transform_sales',
    python_callable=transform_sales,
    provide_context=True,
    dag=dag,
)

load_task = BashOperator(
    task_id='load_to_warehouse',
    bash_command='echo "Loading to warehouse... row count: {{ ti.xcom_pull(key=\"transformed_data\") | length }}" && echo "Load complete"',
    dag=dag,
)

notify_task = BashOperator(
    task_id='send_notification',
    bash_command='echo "Sales pipeline completed at $(date)"',
    dag=dag,
)

# Dependencies
extract_task >> validate_task >> transform_task >> load_task >> notify_task

# airflow_simulator.py
# Simulate Airflow DAG execution locally (no Airflow needed)
from datetime import datetime

class TaskInstance:
    def __init__(self):
        self.xcom_data = {}

    def xcom_push(self, key, value):
        self.xcom_data[key] = value

    def xcom_pull(self, task_ids, key):
        return self.xcom_data.get(key)

class DAGSimulator:
    def __init__(self, dag_id):
        self.dag_id = dag_id
        self.tasks = []

    def run(self):
        print(f"╔══ Running DAG: {self.dag_id} ══╗\n")
        context = {"task_instance": TaskInstance(), "execution_date": datetime.now()}

        results = []
        for task_name, task_func in self.tasks:
            print(f"▶ Task: {task_name}")
            try:
                task_func(**context)
                print(f"  ✓ {task_name} succeeded\n")
                results.append({"task": task_name, "status": "success"})
            except Exception as e:
                print(f"  ✗ {task_name} FAILED: {e}\n")
                results.append({"task": task_name, "status": "failed"})

        print("╚══ DAG Complete ═══════════════╝")
        return results

# Build and run
sim = DAGSimulator("sales_etl_pipeline")
sim.tasks = [
    ("extract_sales", extract_sales),
    ("validate_sales", validate_sales),
    ("transform_sales", transform_sales),
]
results = sim.run()

print("\nExecution Summary:")
for r in results:
    print(f"  {r['task']}: {r['status']}")

# file_sensor.py
# Simulate an Airflow file sensor that waits for a file to appear
import time
import os
from datetime import datetime

class FileSensor:
    def __init__(self, filepath, timeout=60, poke_interval=5):
        self.filepath = filepath
        self.timeout = timeout
        self.poke_interval = poke_interval

    def poke(self):
        """Check if the file exists."""
        return os.path.exists(self.filepath)

    def run(self):
        print(f"[SENSOR] Waiting for {self.filepath}...")
        start = time.time()
        while time.time() - start < self.timeout:
            if self.poke():
                print(f"[SENSOR] File found after {time.time() - start:.1f}s")
                return True
            print(f"[SENSOR] Not yet... checking again in {self.poke_interval}s")
            time.sleep(self.poke_interval)
        print(f"[SENSOR] TIMEOUT after {self.timeout}s")
        return False

# Simulate file appearing after some time
import threading
def delayed_file_creation(path, delay=8):
    time.sleep(delay)
    with open(path, 'w') as f:
        f.write("ready")
    print(f"[SIM] Created file: {path}")

watch_file = "/tmp/dodatech_ready.txt"
threading.Thread(target=delayed_file_creation, args=(watch_file,), daemon=True).start()

sensor = FileSensor(watch_file, timeout=20, poke_interval=3)
result = sensor.run()
print(f"\nSensor result: {'SUCCESS' if result else 'FAILURE'}")

# Cleanup
if os.path.exists(watch_file):
    os.remove(watch_file)