Automated ETL Pipeline

Create a Python CLI tool called etl-pipeline that extracts data from multiple
source formats, transforms it using SQL logic, and loads it into a SQLite
database. Use Python 3.10+ with these packages: pandas, pyyaml, requests,
jinja2. Structure the project properly.

PROJECT STRUCTURE:
etl_pipeline/
├── etl_pipeline/
│   ├── __init__.py
│   ├── __main__.py             # enables python -m etl_pipeline
│   ├── cli.py                  # argparse CLI entry point
│   ├── extractor.py            # reads CSV, JSON, and API sources
│   ├── transformer.py          # SQL transformations in SQLite
│   ├── loader.py               # loads data into target SQLite database
│   ├── quality_report.py       # generates HTML data quality report
│   ├── config.py               # YAML config loader and validator
│   └── templates/
│       └── report.html         # Jinja2 template for data quality report
├── config.yaml                 # default pipeline configuration
├── sample_data/
│   ├── orders.csv              # e-commerce orders (100 rows)
│   ├── customers.json          # customer records (30 records)
│   └── products.csv            # product catalog (20 rows)
├── requirements.txt
├── setup.py
└── README.md

CLI INTERFACE:
  python -m etl_pipeline --config config.yaml --output warehouse.db
  python -m etl_pipeline --config config.yaml --output warehouse.db --report quality_report.html
  python -m etl_pipeline --config config.yaml --output warehouse.db --dry-run
  python -m etl_pipeline --config config.yaml --output warehouse.db --verbose

OPTIONS:
  --config, -c       Path to YAML config file (required)
  --output, -o       Path to output SQLite database (required)
  --report, -r       Path to output HTML data quality report (optional)
  --dry-run          Validate config and sources without writing to the database
  --verbose          Enable debug logging with row counts at each stage
  --force            Overwrite existing database without confirmation

SAMPLE DATA:

1. orders.csv (100 rows):
   Generate a realistic e-commerce orders dataset with columns:
   - order_id (integer, unique, 1001-1100)
   - customer_id (integer, references customers, 1-30)
   - product_id (integer, references products, 1-20)
   - order_date (date, between 2024-01-01 and 2024-12-31)
   - quantity (integer, 1-10)
   - unit_price (decimal, should match the product's price)
   - discount_pct (decimal, 0-25, mostly 0)
   - status (string: "completed", "shipped", "pending", "returned")
   Include realistic patterns: some customers order frequently, some products
   are more popular, Q4 has more orders, some duplicate order_ids to test
   deduplication (5 exact duplicates).

2. customers.json (30 records):
   Generate as a JSON array of objects with:
   - id (integer, 1-30)
   - first_name, last_name (realistic names)
   - email (realistic email addresses)
   - region (string: "North", "South", "East", "West")
   - signup_date (date, between 2022-01-01 and 2024-06-30)
   - tier (string: "bronze", "silver", "gold", "platinum")
   Include some records with missing email (null) and inconsistent region
   capitalization ("north" vs "North") to test data cleaning.

3. products.csv (20 rows):
   - product_id (integer, 1-20)
   - product_name (realistic product names: electronics, office supplies, etc.)
   - category (string: "Electronics", "Furniture", "Office Supplies", "Software")
   - base_price (decimal, 9.99-999.99)
   - supplier (string, 5 different supplier names)
   - in_stock (boolean, 1 or 0)

EXTRACTOR (extractor.py):
  Extract data from each source type into pandas DataFrames:

  1. CSV extraction:
     - Read CSV with configurable encoding (default UTF-8), delimiter, and
       quoting rules
     - Auto-detect date columns and parse them
     - Handle common null representations: "", "N/A", "null", "NULL", "-"
     - Return DataFrame with source metadata (filename, row count, columns)

  2. JSON extraction:
     - Read JSON arrays or JSON objects with a configurable root key
     - Flatten nested objects one level deep (e.g., address.city -> address_city)
     - Handle both .json files and .jsonl (line-delimited) formats
     - Return DataFrame with source metadata

  3. API extraction (stub):
     - Accept a URL endpoint and optional headers/params from config
     - If the URL is a file:// path, read the local file instead (for testing)
     - Implement retry logic with 3 attempts and exponential backoff
     - Return DataFrame with source metadata

  Each extractor logs: source path, row count, column count, detected types.

TRANSFORMER (transformer.py):
  Load all extracted DataFrames into a temporary SQLite database and execute
  SQL transformations defined in the config:

  1. STAGING: Load each DataFrame into a staging table named after the source
     (e.g., stg_orders, stg_customers, stg_products).

  2. CLEANING:
     - Standardize text columns: strip whitespace, consistent capitalization
       per config (upper, lower, title)
     - Coerce types: convert string dates to DATE, string numbers to NUMERIC
     - Replace configured null synonyms with actual NULL
     - Log cleaning actions: "Standardized 5 region values to title case"

  3. DEDUPLICATION:
     - For each table, apply dedup rules from config
     - Support DISTINCT (exact row dedup) and window function dedup:
       ROW_NUMBER() OVER (PARTITION BY key_cols ORDER BY sort_col DESC)
       keeping only row_number = 1
     - Log: "Removed 5 duplicate rows from stg_orders by order_id"

  4. JOINS:
     - Execute JOIN queries defined in config to create merged tables
     - Example: JOIN stg_orders with stg_customers ON customer_id,
       JOIN result with stg_products ON product_id
     - Support INNER JOIN, LEFT JOIN
     - Create a merged table: order_details with all columns from all 3 sources

  5. AGGREGATIONS:
     - Execute GROUP BY queries defined in config
     - Example: revenue_by_region = SELECT region, SUM(quantity * unit_price)
       as total_revenue, COUNT(*) as order_count FROM order_details
       GROUP BY region
     - Example: customer_summary = SELECT customer_id, first_name, last_name,
       COUNT(*) as total_orders, SUM(quantity * unit_price) as lifetime_value
       FROM order_details GROUP BY customer_id, first_name, last_name

  Each transformation step logs its SQL query and resulting row count.

LOADER (loader.py):
  Load transformed tables into the target SQLite database:

  1. Schema validation:
     - Before loading, verify that the transformed data matches expected
       column names and types defined in config
     - Log mismatches as warnings, fail on critical mismatches (missing
       required columns)

  2. Table creation:
     - CREATE TABLE IF NOT EXISTS with proper column types and constraints
     - Add PRIMARY KEY, NOT NULL, and FOREIGN KEY constraints from config
     - Create indexes on foreign key columns and configured index columns

  3. Data loading:
     - INSERT OR REPLACE (upsert) for idempotent loading
     - Batch inserts (500 rows per batch) with transaction wrapping
     - Enable PRAGMA foreign_keys = ON for FK enforcement

  4. Post-load verification:
     - SELECT COUNT(*) from each loaded table
     - Verify foreign key integrity: no orphaned references
     - Log final row counts and any integrity issues

QUALITY REPORT (quality_report.py):
  Generate an HTML data quality report using the Jinja2 template:

  1. SOURCE SUMMARY:
     - Table showing each source file: name, format, row count, column count
     - Total rows extracted vs total rows loaded (with loss percentage)

  2. NULL ANALYSIS:
     - Per-column null counts and percentages for each table
     - Color-coded: green (0%), yellow (1-10%), red (>10%)
     - Highlight columns where nulls exceed a configurable threshold

  3. DUPLICATE ANALYSIS:
     - Exact duplicate counts per table (before and after dedup)
     - Near-duplicate counts by configured key columns
     - Show sample duplicate rows for manual review

  4. TYPE MISMATCH ANALYSIS:
     - Columns where values do not match expected types
     - Show sample mismatched values (first 5)
     - Suggest type coercions for common mismatches

  5. REFERENTIAL INTEGRITY:
     - Foreign key violations: child rows with no matching parent
     - Show counts and sample orphaned records

  6. PIPELINE SUMMARY:
     - Total execution time
     - Steps completed with pass/warn/fail status
     - Overall pipeline health: PASS (all green), WARNING (some yellow),
       FAIL (any red)

  STYLING:
  - Professional report matching the corporate theme from Lesson 4:
    navy headers, white background, gray accents, professional fonts
  - Tables with alternating row colors and formatted numbers
  - Status badges: green PASS, yellow WARNING, red FAIL
  - Print-friendly CSS

CONFIG (config.yaml):
  pipeline:
    name: "E-Commerce Data Warehouse"
    description: "Merge orders, customers, and products into a unified warehouse"

  sources:
    orders:
      type: csv
      path: "sample_data/orders.csv"
      encoding: "utf-8"
      delimiter: ","
    customers:
      type: json
      path: "sample_data/customers.json"
      root_key: null  # top-level array
    products:
      type: csv
      path: "sample_data/products.csv"
      encoding: "utf-8"

  cleaning:
    null_synonyms: ["N/A", "n/a", "NA", "", "null", "NULL", "-"]
    text_standardization:
      - column: "region"
        case: "title"  # North, South, East, West
      - column: "status"
        case: "lower"
      - column: "tier"
        case: "lower"

  deduplication:
    orders:
      method: "window"
      partition_by: ["order_id"]
      order_by: "order_date"
      order_direction: "desc"
    customers:
      method: "distinct"
    products:
      method: "distinct"

  transforms:
    - name: "order_details"
      type: "join"
      sql: |
        SELECT
          o.order_id, o.order_date, o.quantity, o.unit_price, o.discount_pct,
          o.status,
          c.first_name, c.last_name, c.email, c.region, c.tier,
          p.product_name, p.category, p.base_price, p.supplier,
          (o.quantity * o.unit_price * (1 - o.discount_pct / 100.0)) as line_total
        FROM stg_orders o
        LEFT JOIN stg_customers c ON o.customer_id = c.id
        LEFT JOIN stg_products p ON o.product_id = p.product_id

    - name: "revenue_by_region"
      type: "aggregation"
      sql: |
        SELECT
          region,
          COUNT(*) as order_count,
          SUM(line_total) as total_revenue,
          AVG(line_total) as avg_order_value,
          COUNT(DISTINCT order_id) as unique_orders
        FROM order_details
        GROUP BY region
        ORDER BY total_revenue DESC

    - name: "customer_summary"
      type: "aggregation"
      sql: |
        SELECT
          first_name || ' ' || last_name as customer_name,
          email, region, tier,
          COUNT(*) as total_orders,
          SUM(line_total) as lifetime_value,
          AVG(line_total) as avg_order_value,
          MIN(order_date) as first_order,
          MAX(order_date) as last_order
        FROM order_details
        GROUP BY first_name, last_name, email, region, tier
        ORDER BY lifetime_value DESC

    - name: "category_performance"
      type: "aggregation"
      sql: |
        SELECT
          category,
          COUNT(*) as total_orders,
          SUM(quantity) as total_units,
          SUM(line_total) as total_revenue,
          AVG(line_total) as avg_order_value
        FROM order_details
        GROUP BY category
        ORDER BY total_revenue DESC

  load:
    tables:
      - name: "order_details"
        primary_key: "order_id"
        indexes: ["region", "category", "order_date"]
      - name: "revenue_by_region"
        primary_key: "region"
      - name: "customer_summary"
        primary_key: "customer_name"
        indexes: ["region", "tier"]
      - name: "category_performance"
        primary_key: "category"

  quality:
    null_threshold: 10  # percentage — flag columns above this
    duplicate_check_keys:
      orders: ["order_id"]
      customers: ["email"]
      products: ["product_id"]

Generate all files with complete, working implementations. Include the sample
data with realistic values and deliberate quality issues (duplicates, nulls,
inconsistent casing). The tool should run end-to-end on the first try.

[EXTRACT] orders.csv: 105 rows, 8 columns (CSV)
[EXTRACT] customers.json: 30 rows, 7 columns (JSON)
[EXTRACT] products.csv: 20 rows, 6 columns (CSV)
[CLEAN] Standardized 4 region values to title case in stg_customers
[CLEAN] Standardized 2 status values to lower case in stg_orders
[CLEAN] Replaced 3 null synonyms with NULL in stg_customers.email
[DEDUP] Removed 5 duplicate rows from stg_orders (window: order_id, order_date desc)
[DEDUP] 0 duplicates found in stg_customers (distinct)
[DEDUP] 0 duplicates found in stg_products (distinct)
[TRANSFORM] order_details: 100 rows (JOIN orders + customers + products)
[TRANSFORM] revenue_by_region: 4 rows (GROUP BY region)
[TRANSFORM] customer_summary: 30 rows (GROUP BY customer)
[TRANSFORM] category_performance: 4 rows (GROUP BY category)
[LOAD] order_details: 100 rows loaded (upsert)
[LOAD] revenue_by_region: 4 rows loaded (upsert)
[LOAD] customer_summary: 30 rows loaded (upsert)
[LOAD] category_performance: 4 rows loaded (upsert)
[VERIFY] Foreign key integrity: PASS
[REPORT] Data quality report written to quality_report.html
[DONE] Pipeline completed in 1.2s — 4 tables, 138 rows loaded

.tables
-- order_details  revenue_by_region  customer_summary  category_performance

SELECT * FROM revenue_by_region;
-- North|28|12450.50|444.66|25
-- South|24|10230.75|426.28|22
-- East|26|11890.20|457.31|24
-- West|22|9870.40|448.65|20

SELECT customer_name, total_orders, lifetime_value
FROM customer_summary ORDER BY lifetime_value DESC LIMIT 5;
-- (Top 5 customers by lifetime value)

.quit

Add incremental (delta) loading support. Instead of reloading the entire dataset
every run, track a high-water mark (the maximum value of a configured column,
like order_date or order_id) in a metadata table in the target database. On the
next run, only extract rows where the tracked column is greater than the stored
high-water mark. Update the high-water mark after a successful load. This turns
the pipeline from a full refresh to an append-only incremental load. Add a
--full-refresh flag that ignores the high-water mark and reloads everything.

Add data lineage metadata to every loaded row. For each row in the target
tables, add columns: _source_file (which file the row came from),
_extracted_at (timestamp of extraction), _pipeline_run_id (UUID for this run),
and _transform_steps (comma-separated list of transforms applied). Store a
pipeline_runs metadata table with run_id, start_time, end_time, status,
config_hash, and row counts. This creates an audit trail showing exactly where
every row came from and when.

Add automatic schema migration. When the config defines a new column that does
not exist in the target database, automatically run ALTER TABLE ADD COLUMN
instead of failing. When a column type changes, log a warning and create a
migration script (saved as a .sql file) that the user can review and run
manually. Store a schema_versions table that tracks all migrations with
timestamps.

Add a --target flag that accepts "sqlite" or "postgresql". When PostgreSQL is
selected, use psycopg2 to connect to a PostgreSQL database specified by a
connection string in the config. Adjust SQL syntax: use ON CONFLICT for upsert
instead of INSERT OR REPLACE, use PostgreSQL data types (SERIAL, TIMESTAMPTZ),
and use the COPY command for bulk loading instead of batch inserts. Keep SQLite
as the default for local development.