Mage AI

Executes Python, SQL, and R code blocks as nodes in a directed acyclic graph (DAG), where each block is a discrete, reusable unit with explicit input/output dependencies. The execution engine respects block ordering based on data dependencies, manages variable state between blocks via a shared context, and supports both interactive notebook-style development and production-grade pipeline runs. Blocks can be edited interactively with real-time execution feedback, then promoted to scheduled pipelines without code refactoring.

Generates Python, SQL, and R code templates for data loading, transformation, and export blocks using integrated LLM capabilities. The system prompts users for intent (e.g., 'load CSV from S3', 'deduplicate records'), then generates boilerplate code that can be edited interactively. Generated code includes error handling, logging, and type hints. The LLM context includes available data sources, schema information, and pipeline history to produce contextually relevant code.

Automatically detects data dependencies between blocks by analyzing variable references and generates a DAG (directed acyclic graph) without requiring explicit dependency declarations. When a block reads a variable produced by another block, Mage infers the dependency and enforces execution order. The system detects circular dependencies and prevents execution. Dynamic DAGs allow conditional execution: blocks can be skipped based on upstream results or runtime conditions. Dependency visualization shows the pipeline structure graphically, helping users understand data flow.

Executes SQL queries directly against connected databases (PostgreSQL, Snowflake, BigQuery, etc.) without materializing results to Python. The SQL execution engine (SQL Block Execution subsystem) sends queries to the database, retrieves results, and optionally materializes them as DataFrames. Supports parameterized queries to prevent SQL injection, transaction management (commit/rollback), and query profiling (execution time, rows affected). Results can be stored as temporary tables or views for use by downstream blocks. The system detects the database type and applies dialect-specific optimizations.

Tracks pipeline execution metrics (duration, success/failure, resource usage) and sends alerts on failures, timeouts, or SLA violations. The monitoring system stores execution history in a persistent database, enabling trend analysis and performance debugging. Alerts can be configured per-pipeline (email, Slack, PagerDuty, webhooks) and include execution logs and error details. SLA tracking monitors whether pipelines complete within expected time windows; violations trigger alerts. The system provides dashboards showing pipeline health, execution trends, and failure rates.

Processes data incrementally by tracking which records have been processed and only processing new/changed records in subsequent runs. The checkpoint system stores metadata (last processed timestamp, record IDs, hashes) in external storage (database, S3). Blocks can query the checkpoint to determine which records to process. The system supports multiple incremental strategies: timestamp-based (process records after last run), change-data-capture (CDC), and hash-based (process records with changed values). Checkpoints are versioned and can be reset for backfill.

Manages connections to 50+ data sources (databases, data warehouses, APIs, cloud storage) through a centralized io_config.yaml configuration file. The I/O system provides a unified interface (mage_ai/io/base.py) that abstracts source-specific connection logic, allowing blocks to reference data sources by name rather than managing credentials directly. Supports credential injection via environment variables, secrets managers, and OAuth flows. Each data source type (Airtable, Postgres, S3, BigQuery, etc.) has a dedicated loader/exporter module with pre-built templates.

Executes pipelines in response to events (file uploads, API webhooks, message queue events) with sub-second latency for streaming data. The trigger system (Triggers and Events subsystem) supports multiple event sources: S3 file uploads, Kafka topics, webhooks, and scheduled intervals. Streaming pipelines process data incrementally, maintaining state between runs via checkpoints. The execution engine batches incoming events and executes pipeline blocks with streaming-optimized memory management to handle continuous data flow without accumulating state.

Schedules pipeline execution using cron expressions, fixed intervals, or event-based triggers (file uploads, webhooks, manual runs). The scheduler (Pipeline Scheduler subsystem) maintains a queue of pending runs, executes them in order, and tracks execution history with logs and metrics. Supports backfill (running pipelines for past date ranges), conditional execution (skip if upstream failed), and retry logic (exponential backoff). Pipeline runs are isolated; each run has its own execution context and variable namespace, preventing state leakage between runs.

Provides a web-based code editor (React frontend, mage_ai/frontend) where users write and execute Python, SQL, and R code blocks with real-time feedback. Each block execution is isolated; variables are stored in a shared context accessible to downstream blocks. The editor supports syntax highlighting, code completion, and inline error messages. Users can inspect variable values, data types, and DataFrame previews without writing print statements. Execution results (stdout, stderr, exceptions) are displayed inline with line-number references.

Validates data quality at block boundaries using schema definitions, null checks, and custom validation rules. The validation system (Data Cleaning subsystem) allows users to define expected data types, column names, value ranges, and uniqueness constraints. Validation runs automatically after block execution; failures can be configured to block downstream execution or log warnings. Supports both schema-based validation (Pydantic, Great Expectations) and custom Python validation functions. Validation results are tracked in execution history for audit and debugging.

Stores pipeline definitions (blocks, connections, schedules) in Git-compatible format, enabling version control, collaboration, and rollback. Each pipeline is represented as a directory with YAML files (metadata.yaml, io_config.yaml) and Python/SQL block files. The system tracks changes, supports branching, and can merge pipeline changes from multiple developers. Automatic conflict resolution uses last-write-wins for non-conflicting changes; conflicting changes require manual resolution. Integration with GitHub, GitLab, and Bitbucket allows CI/CD workflows (e.g., run tests on PR, deploy on merge).

Generates interactive charts and visualizations from block outputs without requiring additional code. The visualization system (Data Visualization subsystem) automatically detects DataFrame structure and suggests appropriate chart types (line, bar, scatter, heatmap, etc.). Users can customize axes, aggregations, and filters through a UI. Visualizations are embedded in the pipeline editor, allowing exploratory analysis alongside code development. Supports both static (matplotlib, seaborn) and interactive (Plotly, Altair) charting libraries.

Deploys pipelines to multiple environments (dev, staging, production) with environment-specific configurations. The deployment system uses environment variables and configuration files to manage differences between environments (database connections, API endpoints, data paths). Pipelines are deployed as Docker containers or directly to cloud platforms (AWS ECS, Google Cloud Run, Kubernetes). The system supports blue-green deployments (running old and new versions in parallel) and canary deployments (gradually rolling out changes). Deployment history and rollback capabilities are built-in.