ETL Pipelines Explained — Extract, Transform, Load with Python Examples

# etl_pipeline.py
# Complete ETL: CSV → Transform → Database (simulated)
import csv
import sqlite3
from datetime import datetime

# ── Extract ──
def extract_csv(filepath):
    print(f"[EXTRACT] Reading {filepath}...")
    with open(filepath, 'r') as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    print(f"[EXTRACT] Found {len(rows)} rows")
    return rows

# ── Transform ──
def transform_sales(rows):
    print("[TRANSFORM] Cleaning and enriching...")
    transformed = []
    errors = 0
    for row in rows:
        try:
            # Parse and standardize date
            row['order_date'] = datetime.strptime(row['order_date'], '%Y-%m-%d').date().isoformat()
            # Clean price (remove $, convert to float)
            row['amount'] = float(row['amount'].replace('$', '').replace(',', ''))
            # Normalize status
            row['status'] = row['status'].strip().lower()
            # Add derived columns
            row['processed_at'] = datetime.now().isoformat()
            if row['amount'] > 100:
                row['category'] = 'high_value'
            else:
                row['category'] = 'standard'
            transformed.append(row)
        except (ValueError, KeyError) as e:
            errors += 1
            print(f"  [WARN] Skipping row: {e}")
    print(f"[TRANSFORM] {len(transformed)} valid, {errors} errors")
    return transformed

# ── Load ──
def load_to_db(rows, db_path):
    print(f"[LOAD] Writing to {db_path}...")
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    cursor.execute('''
        CREATE TABLE IF NOT EXISTS sales (
            order_id TEXT PRIMARY KEY,
            customer_name TEXT,
            order_date TEXT,
            amount REAL,
            status TEXT,
            category TEXT,
            processed_at TEXT
        )
    ''')

    inserted = 0
    for row in rows:
        try:
            cursor.execute('''
                INSERT OR REPLACE INTO sales
                (order_id, customer_name, order_date, amount, status, category, processed_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            ''', (
                row['order_id'], row['customer_name'], row['order_date'],
                row['amount'], row['status'], row['category'], row['processed_at']
            ))
            inserted += 1
        except Exception as e:
            print(f"  [ERROR] Failed to insert {row.get('order_id', 'unknown')}: {e}")

    conn.commit()
    conn.close()
    print(f"[LOAD] Inserted {inserted} rows")

# ── Orchestrate ──
def run_etl():
    print("=" * 50)
    print("ETL Pipeline: Sales Data")
    print(f"Started: {datetime.now().isoformat()}")
    print("=" * 50)

    raw_data = extract_csv('sales_data.csv')
    clean_data = transform_sales(raw_data)
    load_to_db(clean_data, 'sales_warehouse.db')

    print("=" * 50)
    print("ETL Pipeline Complete!")
    print("=" * 50)

# ── Test with sample data ──
def create_sample_csv():
    with open('sales_data.csv', 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(['order_id', 'customer_name', 'order_date', 'amount', 'status'])
        writer.writerow(['1001', 'Alice', '2026-06-01', '$250.00', 'completed'])
        writer.writerow(['1002', 'Bob', '2026-06-02', '$45.50', 'pending'])
        writer.writerow(['1003', 'Charlie', '2026-06-03', '$1,200.00', 'completed'])
        writer.writerow(['1004', 'Diana', '2026-06-04', '$89.99', 'cancelled'])
        writer.writerow(['1005', 'Invalid', 'not-a-date', '$50.00', 'completed'])

if __name__ == '__main__':
    create_sample_csv()
    run_etl()

    # Verify
    print("\n[VERIFY] Querying loaded data:")
    conn = sqlite3.connect('sales_warehouse.db')
    cursor = conn.cursor()
    for row in cursor.execute('SELECT * FROM sales'):
        print(f"  {row}")
    conn.close()

# incremental_etl.py
# Track last processed timestamp for incremental loads

import json
import os

STATE_FILE = 'etl_state.json'

def load_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f)
    return {"last_run": None, "rows_processed": 0}

def save_state(state):
    with open(STATE_FILE, 'w') as f:
        json.dump(state, f, indent=2)

def extract_incremental(last_run):
    print(f"[EXTRACT] Fetching records since {last_run}")
    # Simulate API call with since parameter
    data = [
        {"id": 1, "updated_at": "2026-06-15T08:00:00"},
        {"id": 2, "updated_at": "2026-06-15T09:00:00"},
        {"id": 3, "updated_at": "2026-06-15T10:00:00"},
    ]
    if last_run:
        data = [r for r in data if r["updated_at"] > last_run]
    return data

def transform(rows):
    return [{**r, "processed": True} for r in rows]

def load(rows, cursor):
    for row in rows:
        cursor.append(row)
    return len(rows)

state = load_state()
print(f"Previous state: {state}")

new_data = extract_incremental(state["last_run"])
transformed = transform(new_data)
destination = []
loaded = load(transformed, destination)

state["last_run"] = "2026-06-15T10:00:00"
state["rows_processed"] += loaded
save_state(state)

print(f"Loaded {loaded} new records")
print(f"Current state: {state}")
print(f"Destination has {len(destination)} records")