This document provides a comprehensive overview of Anomstack's architecture, including system design, data flow, and component interactions.
Anomstack is a distributed anomaly detection platform built on modern data stack principles. It combines the orchestration power of Dagster with the simplicity of FastHTML to provide a complete anomaly detection solution.
- Modularity: Clean separation of concerns with pluggable components
- Scalability: Handles everything from single metrics to enterprise-scale deployments
- Flexibility: Support for multiple data sources, ML algorithms, and deployment patterns
- Observability: Built-in monitoring, logging, and visualization
- Simplicity: Minimal configuration required to get started
- Reliability: Direct Python module loading eliminates gRPC network overhead and improves system stability
The system follows a layered architecture with clear separation between data ingestion, processing, storage, and presentation layers.
graph TB
subgraph "Data Sources"
BQ[BigQuery]
SF[Snowflake]
CH[ClickHouse]
DB[DuckDB]
SQ[SQLite]
PY[Python Functions]
end
subgraph "Anomstack Core"
subgraph "Orchestration Layer"
DG[Dagster]
JOBS[Jobs Pipeline]
end
subgraph "Processing Layer"
ING[Ingestion]
PREP[Preprocessing]
ML[ML Training]
SCORE[Scoring]
end
subgraph "Storage Layer"
LOCAL[Local Storage]
S3[Amazon S3]
GCS[Google Cloud Storage]
end
end
subgraph "User Interfaces"
DASH[FastHTML Dashboard]
DGUI[Dagster UI]
API[REST API]
end
subgraph "Alerting System"
EMAIL[Email Alerts]
SLACK[Slack Alerts]
LLM[LLM Agent Alerts]
THRESH[Threshold Alerts]
end
BQ --> ING
SF --> ING
CH --> ING
DB --> ING
SQ --> ING
PY --> ING
ING --> PREP
PREP --> ML
ML --> SCORE
SCORE --> EMAIL
SCORE --> SLACK
SCORE --> LLM
ING --> THRESH
ML --> LOCAL
ML --> S3
ML --> GCS
DG --> JOBS
JOBS --> ING
JOBS --> PREP
JOBS --> ML
JOBS --> SCORE
DASH --> DG
DGUI --> DG
API --> DG
Dagster serves as the orchestration engine, managing the entire anomaly detection pipeline:
- Job Scheduling: Automated execution of metric batches
- Dependency Management: Ensures proper execution order
- Asset Management: Tracks data lineage and freshness
- Monitoring: Built-in observability and alerting
- Web UI: Visual pipeline management and debugging
Anomstack supports multiple data platforms through a unified interface:
- Cloud Data Warehouses: BigQuery, Snowflake, Redshift
- OLAP Databases: ClickHouse, DuckDB
- Traditional Databases: PostgreSQL, SQLite
- Custom Python Functions: For complex data sources
The core processing pipeline consists of several stages:
- Executes SQL queries or Python functions
- Handles parameterization and templating
- Validates data quality and schema
- Evaluates threshold-based alerts at ingest time
- Data cleaning and normalization
- Feature engineering (lags, rolling statistics)
- Time series resampling and aggregation
- Model training using PyOD algorithms
- Hyperparameter optimization
- Model validation and selection
- Real-time anomaly detection
- Confidence interval calculation
- Anomaly threshold evaluation
Built on PyOD (Python Outlier Detection):
- Multiple Algorithms: Isolation Forest, OCSVM, LOF, Autoencoders
- Ensemble Methods: Combine multiple algorithms for better accuracy
- Online Learning: Continuous model updates with new data
- Model Persistence: Versioned model storage and retrieval
Flexible storage architecture supporting multiple backends:
- Local Storage: File system for development and small deployments
- Cloud Storage: S3, GCS for scalable model and data storage
- Database Storage: Direct database integration for metadata
Multi-channel alerting with intelligent filtering:
- Email Alerts: Rich HTML emails with visualizations
- Slack Integration: Real-time team notifications
- LLM Agent Alerts: AI-powered contextual analysis
- Threshold Alerts: Traditional rules-based alerting with configurable upper/lower bounds evaluated at ingest time
- Webhook Support: Custom integrations
The following diagram illustrates the complete data flow through the Anomstack pipeline:
flowchart TD
subgraph "Data Sources"
DS1[BigQuery]
DS2[Snowflake]
DS3[ClickHouse]
DS4[DuckDB]
DS5[SQLite]
DS6[Python Functions]
end
subgraph "Dagster Orchestration"
subgraph "Jobs Pipeline"
J1[Ingest Job]
J2[Train Job]
J3[Score Job]
J4[Alert Job]
J5[Plot Job]
end
subgraph "Schedules & Sensors"
SCH[Schedules]
SEN[Sensors]
end
end
subgraph "Processing Components"
ING[Data Ingestion<br/>SQL/Python Execution]
PREP[Preprocessing<br/>Feature Engineering]
TRAIN[ML Training<br/>PyOD Algorithms]
SCORE[Anomaly Scoring<br/>Real-time Detection]
VIZ[Visualization<br/>Plot Generation]
end
subgraph "Storage Systems"
MS[Model Storage<br/>Local/S3/GCS]
DS[Data Storage<br/>Cache/Database]
META[Metadata<br/>Dagster Assets]
end
subgraph "User Interfaces"
DASH[FastHTML Dashboard<br/>Metrics & Anomalies]
DGUI[Dagster UI<br/>Pipeline Management]
API[REST API<br/>Programmatic Access]
end
subgraph "Alerting Channels"
EMAIL[Email Alerts<br/>SMTP/HTML]
SLACK[Slack Notifications<br/>Webhooks]
LLM[LLM Agent<br/>AI Analysis]
THRESH[Threshold Alerts<br/>Rules-based]
end
%% Data flow connections
DS1 --> J1
DS2 --> J1
DS3 --> J1
DS4 --> J1
DS5 --> J1
DS6 --> J1
J1 --> ING
ING --> PREP
PREP --> J2
J2 --> TRAIN
TRAIN --> MS
MS --> J3
J3 --> SCORE
SCORE --> J4
SCORE --> J5
J4 --> EMAIL
J4 --> SLACK
J4 --> LLM
J1 --> THRESH
J5 --> VIZ
%% Orchestration
SCH --> J1
SCH --> J2
SCH --> J3
SCH --> J4
SEN --> J1
SEN --> J3
%% Storage connections
ING --> DS
PREP --> DS
SCORE --> DS
VIZ --> DS
TRAIN --> META
SCORE --> META
%% UI connections
DASH --> DS
DASH --> META
DGUI --> META
API --> DS
API --> META
sequenceDiagram
participant Config as Metric Config
participant Dagster as Dagster Job
participant SQL as SQL Engine
participant DS as Data Source
participant Cache as Data Cache
Config->>Dagster: Trigger Ingest Job
Dagster->>SQL: Load SQL Template
SQL->>SQL: Parameter Substitution
SQL->>DS: Execute Query
DS->>SQL: Return Results
SQL->>Cache: Store Raw Data
Cache->>Dagster: Confirm Success
sequenceDiagram
participant Cache as Data Cache
participant Prep as Preprocessor
participant ML as ML Engine
participant PyOD as PyOD Library
participant Storage as Model Storage
Cache->>Prep: Load Historical Data
Prep->>Prep: Feature Engineering
Prep->>ML: Processed Dataset
ML->>PyOD: Train Model
PyOD->>ML: Trained Model
ML->>Storage: Persist Model
sequenceDiagram
participant Cache as Data Cache
participant Storage as Model Storage
participant Score as Scoring Engine
participant Alert as Alert System
participant UI as Dashboard
Cache->>Score: New Data Point
Storage->>Score: Load Trained Model
Score->>Score: Calculate Anomaly Score
alt Score > Threshold
Score->>Alert: Trigger Alert
Alert->>Alert: Generate Notifications
end
Score->>UI: Update Dashboard
Anomstack uses a configuration-driven approach where metric batches are defined in YAML files:
# metrics/sales/daily_revenue.yaml
metric_batch: sales
metrics:
- daily_revenue
data_source: bigquery
schedule: "0 9 * * *" # Daily at 9 AM
ml_algorithm: IsolationForest
anomaly_threshold: 0.8
alerts:
email_enabled: true
slack_enabled: trueThe system uses a plugin-based architecture for extensibility:
- Data Source Plugins: Easy integration of new databases
- ML Algorithm Plugins: Support for custom anomaly detection methods
- Alert Channel Plugins: Integration with new notification systems
- Storage Backend Plugins: Support for new storage systems
Anomstack supports multiple deployment patterns to meet different operational requirements:
graph LR
subgraph "Local Machine"
D[Dagster]
DB[(SQLite)]
FS[File System]
DASH[Dashboard]
end
D --> DB
D --> FS
DASH --> D
Simplified 3-container architecture with direct Python module loading (no gRPC server):
graph TB
subgraph "Docker Environment"
subgraph "Consolidated Container"
DGW[Dagster Webserver + User Code<br/>(Direct Python Module Loading)]
end
subgraph "Daemon Container"
DGD[Dagster Daemon]
end
subgraph "Dashboard Container"
DASH[FastHTML Dashboard]
end
subgraph "Storage"
SQLITE[(SQLite Files)]
DUCKDB[(DuckDB Volume)]
end
end
DGW --> SQLITE
DGW --> DUCKDB
DGD --> SQLITE
DGD --> DUCKDB
DASH --> DUCKDB
graph TB
subgraph "Dagster Cloud"
subgraph "Serverless"
JOBS[Jobs]
SCHED[Schedules]
end
subgraph "Branch Deployments"
PROD[Production]
STAGE[Staging]
DEV[Development]
end
end
subgraph "Cloud Services"
BQ[(BigQuery)]
S3[(S3 Storage)]
SLACK[Slack API]
EMAIL[Email Service]
end
JOBS --> BQ
JOBS --> S3
JOBS --> SLACK
JOBS --> EMAIL
By default, Anomstack uses direct Python module loading for simplicity and reliability. However, users can optionally configure separate gRPC code servers for advanced deployment scenarios such as:
- Multi-language environments: When integrating with non-Python services
- Container isolation: When user code needs strict separation from Dagster services
- High-security environments: When code execution requires additional sandboxing
- Distributed deployments: When code servers need to run on separate machines
To enable gRPC code servers, modify your workspace.yaml:
load_from:
- grpc_server:
host: localhost
port: 4000
location_name: "anomstack_code"And update your deployment to include a separate code server:
# docker-compose.yaml example
services:
code_server:
image: anomstack_code:latest
command: dagster code-server start -h 0.0.0.0 -p 4000 -f anomstack/main.py
ports:
- "4000:4000"
dagster_webserver:
image: anomstack_dagster:latest
depends_on:
- code_server
ports:
- "3000:3000"Benefits of gRPC approach:
- Stronger isolation between user code and Dagster services
- Support for non-Python user code
- Better resource allocation control
Benefits of direct module loading (default):
- Simpler deployment and configuration
- Eliminates network overhead and potential connectivity issues
- Faster startup times and more reliable execution
- Easier debugging and development workflow
For most use cases, the default direct Python module loading is recommended.
- API Keys: Secure access to external services
- IAM Roles: Cloud-native identity management
- Environment Variables: Secure configuration management
- VPC/Network Isolation: Network-level security
- Encryption at Rest: Model and data encryption
- Encryption in Transit: TLS for all communications
- Data Anonymization: PII handling in preprocessing
- Audit Logging: Comprehensive activity tracking
Anomstack leverages Dagster's built-in observability features and includes its own monitoring capabilities:
graph TD
subgraph "Dagster Observability"
RUNS[Job Runs & Status]
ASSETS[Asset Lineage]
LOGS[Execution Logs]
SENSORS[Failure Sensors]
end
subgraph "Anomstack Monitoring"
DASH[FastHTML Dashboard<br/>Metrics Visualization]
PLOTS[Plot Jobs<br/>Visual Diagnostics]
ALERTS[Alert System<br/>Email/Slack/LLM]
end
subgraph "External Integration"
DAGUI[Dagster UI<br/>Pipeline Monitoring]
EMAIL[Email Notifications]
SLACK[Slack Notifications]
end
RUNS --> DAGUI
ASSETS --> DAGUI
LOGS --> DAGUI
SENSORS --> ALERTS
DASH --> PLOTS
ALERTS --> EMAIL
ALERTS --> SLACK
- Dagster UI: Built-in pipeline monitoring, job status, and execution logs
- FastHTML Dashboard: Real-time metrics visualization and anomaly tracking
- Plot Jobs: Visual diagnostics embedded in Dagster job logs
- Failure Sensors: Automatic detection of pipeline failures (
anomstack/sensors/failure.py) - Alert System: Multi-channel notifications for anomalies and system issues
- Asset Lineage: Track data dependencies and freshness through Dagster assets
- Dagster Cloud: Serverless scaling for job execution
- Database Sharding: Distribute metrics across multiple databases
- Model Parallelization: Train multiple models concurrently
- Alert Batching: Batch notifications to reduce load
- Data Caching: Redis/Memcached for frequently accessed data
- Model Caching: Avoid reloading models for each prediction
- Query Optimization: Efficient SQL queries with proper indexing
- Lazy Loading: Load data and models only when needed
The architecture provides several extension points:
- Custom Data Sources: Implement new data source connectors
- Custom ML Algorithms: Add new anomaly detection methods
- Custom Preprocessing: Implement domain-specific data transformations
- Custom Alerts: Create new notification channels
- Custom Visualizations: Add new chart types and dashboards
Anomstack provides APIs for external integration:
- REST API: HTTP-based programmatic access
- GraphQL API: Flexible query interface
- Webhook API: Event-driven integrations
- Metrics API: Export metrics to external systems
- Configuration Backup: Version-controlled YAML configurations
- Model Backup: Versioned model storage with retention policies
- Data Backup: Point-in-time recovery for historical data
- Metadata Backup: Dagster run history and asset metadata
- Multi-Region Deployment: Geographic distribution for resilience
- Database Replication: Master-slave configuration for data availability
- Load Balancing: Distribute traffic across multiple instances
- Graceful Degradation: Continue operation with reduced functionality
| Layer | Technology | Purpose |
|---|---|---|
| Orchestration | Dagster | Pipeline orchestration and scheduling (consolidated architecture) |
| Web Framework | FastHTML | Dashboard and API development |
| ML Library | PyOD | Anomaly detection algorithms |
| Data Processing | Pandas, NumPy | Data manipulation and analysis |
| Visualization | Plotly, Matplotlib | Chart generation and visualization |
| Database | SQLite, DuckDB | Metadata and metrics storage (simplified deployment) |
| Cloud Storage | S3, GCS | Model and data storage |
| Containerization | Docker | Application packaging and deployment (3-container architecture) |
| Code Loading | Direct Python Modules | User code integration (no gRPC server needed) |
This architecture enables Anomstack to provide a robust, scalable, and maintainable anomaly detection platform that can adapt to various deployment scenarios and organizational requirements.