Skip to content

Analytics Warehouse Operations Guide

Note: The ETL pipeline is now an independent microservice located at src/pipelines/analytics_etl/.
See the ETL Pipeline README for architecture and development details.

Overview

The analytics warehouse provides 10-100x faster query performance through pre-aggregated data stored in MongoDB dimensional tables.

Key Benefits: - 8-10x query speedup (confirmed: 8.8x in production testing) - Dimensional modeling for flexible analytics - Incremental ETL with checkpoint tracking - Deduplication and data quality controls - Backward compatible with existing APIs

Architecture

Database Structure

Warehouse Database: {prefix}_analytics (e.g., craigh_analytics)

MongoDB Logs (event='<prefix>_api') 
    ↓
[Extract] - Only API events, checkpoint tracking
    ↓
[Transform] - Validation, cleaning, deduplication
    ↓
[Load] - Upsert dimensions, insert facts
    ↓
Analytics Warehouse (fact + dimension tables)
    ↓
[Aggregate] - Hourly/daily summaries
    ↓
Fast API Queries (30-50ms vs 300-500ms)

Collections

Fact Tables: - fact_interaction - Atomic grain (1 row per log) - Embedded dimension objects - Time dimensions (ts, date, year, month, week, day, hour) - Measures (response_time_ms, token_counts, cost_usd) - fact_hourly_summary - Pre-aggregated hourly metrics - fact_daily_summary - Pre-aggregated daily metrics

Dimension Tables: - dim_cat - Catalog/CAT information - dim_topic - Topics/categories - dim_model - AI models (gpt-4o, claude-3, etc.) - dim_persona - User personas - dim_question - Unique questions (deduplicated) - dim_answer - Unique answers (deduplicated) - dim_session - User sessions - dim_user - User information

Metadata Tables: - etl_log - ETL job execution history - etl_checkpoint - Last processed timestamp per CAT

Daily Operations

1. Run Incremental ETL

Extract new logs since last checkpoint:

pants run src/pipelines/analytics_etl/scripts:run_etl -- craigh

What it does: - Checks checkpoint for last processed timestamp - Extracts new logs where event='craigh_api' and ts > checkpoint - Transforms and validates data - Loads into warehouse (upserts dimensions, inserts facts) - Updates checkpoint for next run - Logs execution to etl_log collection

Expected output: 

======================================================================
ETL PIPELINE - CRAIGH
======================================================================
Running INCREMENTAL ETL (processing new logs only)...
✓ Extracted 45 logs in 0.32s
✓ Transformed 45 logs in 0.15s
✓ Loaded 45 logs, skipped 0 duplicates, 0 errors in 1.23s
✓ Updated checkpoint to: 1737399845.123
======================================================================
  Extracted:    45
  Transformed:  45
  Loaded:       45
  Skipped:      0 (duplicates)
  Errors:       0
======================================================================

2. Run Aggregations

Update summary tables:

pants run src/pipelines/analytics_etl/scripts:run_aggregation -- craigh

What it does: - Queries fact_interaction for new/updated data - Aggregates by hour and day - Upserts into fact_hourly_summary and fact_daily_summary - Creates indexes for fast queries

Expected output: 

================================================================================
ANALYTICS AGGREGATION: craigh
================================================================================
✓ Initialized aggregator for cat 'craigh'

Running aggregations...

Hourly Summary:
  Processed: 24 hours
  Created: 18 new records
  Updated: 6 existing records

Daily Summary:
  Processed: 7 days
  Created: 5 new records
  Updated: 2 existing records

⏱️  Total duration: 2.34s
================================================================================

3. Check Warehouse Status

Verify data is loaded correctly:

pants run src/pipelines/analytics_etl/scripts:check_warehouse_status -- craigh

Or inspect manually:

pants run src/pipelines/analytics_etl/scripts:inspect_warehouse_data -- craigh

Scheduling (Production)

Cron Setup

Recommended schedule:

# Edit crontab
crontab -e

# Run incremental ETL every hour (at minute 0)
0 * * * * cd /path/to/bast-ai && pants run src/pipelines/analytics_etl/scripts:run_etl -- craigh >> /var/log/bast-etl-craigh.log 2>&1

# Run aggregations 5 minutes after ETL (at minute 5)
5 * * * * cd /path/to/bast-ai && pants run src/pipelines/analytics_etl/scripts:run_aggregation -- craigh >> /var/log/bast-agg-craigh.log 2>&1

For multiple CATs:

# Stagger runs to avoid resource contention
0 * * * * cd /path/to/bast-ai && pants run src/pipelines/analytics_etl/scripts:run_etl -- craigh >> /var/log/bast-etl-craigh.log 2>&1
5 * * * * cd /path/to/bast-ai && pants run src/pipelines/analytics_etl/scripts:run_aggregation -- craigh >> /var/log/bast-agg-craigh.log 2>&1

10 * * * * cd /path/to/bast-ai && pants run src/pipelines/analytics_etl/scripts:run_etl -- advisoruat >> /var/log/bast-etl-advisor.log 2>&1
15 * * * * cd /path/to/bast-ai && pants run src/pipelines/analytics_etl/scripts:run_aggregation -- advisoruat >> /var/log/bast-agg-advisor.log 2>&1

Systemd Timer (Alternative)

# Create timer file
sudo nano /etc/systemd/system/bast-etl-craigh.timer

[Unit]
Description=Bast ETL for craigh (hourly)

[Timer]
OnCalendar=hourly
Persistent=true

[Install]
WantedBy=timers.target

Full Rebuild (Rarely Needed)

When to Use

  • Initial warehouse setup
  • Data corruption recovery
  • Schema changes requiring reload
  • Major data quality issues

How to Run

# Step 1: Backup current warehouse (optional but recommended)
mongodump --db craigh_analytics --out /backup/craigh_analytics_$(date +%Y%m%d)

# Step 2: Clear warehouse (optional - deduplication handles this)
# mongo craigh_analytics --eval "db.dropDatabase()"

# Step 3: Run full ETL (processes all logs, skips duplicates)
pants run src/pipelines/analytics_etl/scripts:run_etl -- craigh --full

# Step 4: Run aggregations
pants run src/pipelines/analytics_etl/scripts:run_aggregation -- craigh

# Step 5: Verify
pants run src/apis/analytics/scripts:inspect_warehouse_data -- craigh

Testing & Validation

Compare Endpoints

Test performance and data accuracy:

# Terminal 1: Start API server
pants run src/apis/analytics/integration:api_tester -- craigh

# Terminal 2: Run comparison (wait for server to start)
PANTS_CONCURRENT=True pants run src/pipelines/analytics_etl/scripts:compare_endpoints -- craigh

Expected results: 

Performance Comparison:
  Legacy:      415.04ms
  Warehouse:    47.17ms
  Speedup:       8.80x faster

Data Comparison:
  Total Interactions: ✓ Match (1,143 = 1,143)
  Topic Counts:       ✓ Match (29 = 29)
  Model Counts:       ✓ Match (11 = 11)
  Persona Counts:     ✓ Match (6 = 6)
  Questions:          ✓ Match (705 = 705)
  Answers:            ✓ Match (1,093 = 1,093)

Manual API Testing

# High-performance warehouse endpoint
curl "http://localhost:9090/api/v1/analytics/all_warehouse?catname=craigh" | jq .

# Original endpoint (for comparison)
curl "http://localhost:9090/api/v1/analytics/all?catname=craigh" | jq .

# Test with filters
curl "http://localhost:9090/api/v1/analytics/all_warehouse?catname=craigh&startDate=2026-01-01&endDate=2026-12-31&model=gpt-4o" | jq .

Monitoring

Check ETL Status

MongoDB queries:

// Recent ETL jobs
use craigh_analytics
db.etl_log.find().sort({job_start_time: -1}).limit(10).pretty()

// Failed jobs
db.etl_log.find({status: "failed"}).pretty()

// Current checkpoint
db.etl_checkpoint.find({cat_name: "craigh"}).pretty()

Check Data Counts

// Fact interaction count
db.fact_interaction.count()

// Recent interactions (last 24 hours)
db.fact_interaction.count({
  ts: {$gte: (Date.now() / 1000) - 86400}
})

// Dimension counts
db.dim_topic.count()
db.dim_model.count()
db.dim_user.count()

// Summary table counts
db.fact_hourly_summary.count()
db.fact_daily_summary.count()

Performance Monitoring

Check API response times:

# Monitor in real-time
while true; do
  curl -s -w "\nTime: %{time_total}s\n" \
    "http://localhost:9090/api/v1/analytics/all_warehouse?catname=craigh" \
    | jq '.performance_ms'
  sleep 5
done

Expected values: - Warehouse: 30-50ms (fast) - Legacy: 300-500ms (slow)

Troubleshooting

ETL Job Failed

Symptoms: - Cron job exits with error - No new data in warehouse - ETL log shows failed status

Diagnosis:

# Check ETL logs
tail -f /var/log/bast-etl-craigh.log

# Check MongoDB ETL log
mongo craigh_analytics --eval "db.etl_log.find({status: 'failed'}).sort({job_start_time: -1}).limit(1).pretty()"

Common causes: 1. MongoDB connection issues 2. Invalid data in source logs 3. Disk space full 4. Permission issues

Fix: 

# 1. Verify MongoDB connection
mongo --eval "db.runCommand({ping: 1})"

# 2. Check disk space
df -h

# 3. Re-run manually with verbose logging
pants run src/pipelines/analytics_etl/scripts:run_etl -- craigh

Missing Data

Symptoms: - Dashboard shows no recent data - API returns old data - Warehouse counts don't match expectations

Diagnosis:

// Check checkpoint
db.etl_checkpoint.find({cat_name: "craigh"})

// Check latest fact
db.fact_interaction.find().sort({ts: -1}).limit(1).pretty()

// Check ETL log
db.etl_log.find().sort({job_start_time: -1}).limit(5).pretty()

Fix: 

# Force re-run (updates checkpoint)
pants run src/pipelines/analytics_etl/scripts:run_etl -- craigh

# If still missing, run full ETL
pants run src/pipelines/analytics_etl/scripts:run_etl -- craigh --full

Data Accuracy Issues

Symptoms: - Counts don't match between legacy and warehouse - Missing topics/models/personas - Incorrect aggregations

Diagnosis:

# Compare endpoints
PANTS_CONCURRENT=True pants run src/pipelines/analytics_etl/scripts:compare_endpoints -- craigh

# Inspect warehouse data
pants run src/apis/analytics/scripts:inspect_warehouse_data -- craigh

Common causes: 1. Event filtering (only event='<prefix>_api' is processed) 2. Deduplication (duplicates are skipped) 3. Missing persona/topic labels in source data 4. Nested field access issues (should use model.name not model_name)

Fix: 

# Re-run full ETL to rebuild warehouse
pants run src/pipelines/analytics_etl/scripts:run_etl -- craigh --full
pants run src/pipelines/analytics_etl/scripts:run_aggregation -- craigh

Slow Query Performance

Symptoms: - Warehouse queries take > 100ms - API timeouts - Dashboard slow to load

Diagnosis:

// Check if indexes exist
db.fact_interaction.getIndexes()
db.fact_hourly_summary.getIndexes()

// Run explain on slow query
db.fact_interaction.find({ts: {$gte: 1737000000}}).explain("executionStats")

Fix:

// Create missing indexes
db.fact_interaction.createIndex({ts: 1})
db.fact_interaction.createIndex({"cat._id": 1, ts: 1})
db.fact_interaction.createIndex({"model.name": 1})
db.fact_interaction.createIndex({"topic.label": 1})

// For aggregations
db.fact_hourly_summary.createIndex({hour_bucket: 1})
db.fact_daily_summary.createIndex({day: 1})

Best Practices

ETL Operations

  1. Run incremental ETL frequently (hourly recommended)
  2. Run aggregations after ETL (5-10 minute delay)
  3. Monitor ETL logs for failures
  4. Backup before major changes (schema updates, full rebuilds)
  5. Test in dev before production

Data Quality

  1. Validate source data before loading
  2. Check for duplicates (source_log_id uniqueness)
  3. Monitor null values (especially topics/personas)
  4. Verify event filtering (only API events processed)
  5. Compare with legacy endpoint after changes

Performance

  1. Create indexes on frequently queried fields
  2. Use aggregation summaries for dashboards
  3. Cache API responses (5-15 minute TTL)
  4. Monitor query times (alert if > 100ms)
  5. Partition large tables if needed (future)

Migration Strategy

Phase 1: Deploy Warehouse (Current)

  • ✅ Deploy warehouse code
  • ✅ /all endpoint unchanged (backward compatible)
  • ✅ /all_warehouse available for testing
  • ✅ Run ETL to populate warehouse

Phase 2: Validation (1-2 weeks)

  • ✅ Run comparison tool regularly
  • ✅ Monitor performance metrics
  • ✅ Validate data accuracy
  • ✅ Test with production traffic

Phase 3: Gradual Migration (TBD)

Option A: Update clients gradually - Dashboard 1: Switch to /all_warehouse - Dashboard 2: Switch to /all_warehouse - Monitor for issues - Eventually deprecate /all_legacy

Option B: Swap implementation - Update /all to use warehouse logic - Keep /all_legacy as fallback - No client changes needed!

Phase 4: Cleanup (Future)

  • Remove /all_legacy endpoint
  • Archive old documentation
  • Optimize warehouse schema
  • Add new features (real-time streaming, etc.)

Support

Documentation: - Performance Comparison - API Endpoints Reference - ETL Design Document

Commands Reference: 

# ETL
pants run src/pipelines/analytics_etl/scripts:run_etl -- <catname> [--full]
pants run src/pipelines/analytics_etl/scripts:run_aggregation -- <catname>

# Testing
pants run src/apis/analytics/integration:api_tester -- <catname>
pants run src/pipelines/analytics_etl/scripts:compare_endpoints -- <catname>

# Inspection
pants run src/pipelines/analytics_etl/scripts:check_warehouse_status -- <catname>
pants run src/pipelines/analytics_etl/scripts:inspect_warehouse_data -- <catname>

Questions? Check the troubleshooting section above or review the ETL logs.