Workflow:Mage ai Mage ai SQL Database Source Extraction
| Knowledge Sources | |
|---|---|
| Domains | Data_Engineering, ETL, Database_Integration |
| Last Updated | 2026-02-09 07:00 GMT |
Overview
End-to-end process for extracting data from relational databases (PostgreSQL, MySQL, BigQuery, Snowflake, Redshift, MSSQL, OracleDB, and others) using the Singer tap protocol with automatic schema discovery and incremental sync support.
Description
This workflow covers the complete lifecycle of a Singer-based SQL source connector in the Mage integrations framework. It leverages the SQL base class to provide automatic schema discovery via information_schema queries, paginated data extraction with configurable batch sizes, and incremental sync through bookmark-based state tracking. The process handles type mapping from SQL column types to JSON Schema types, supports multiple replication methods (FULL_TABLE, INCREMENTAL, LOG_BASED), and emits Singer-protocol messages (SCHEMA, RECORD, STATE) for downstream consumption by any Singer-compatible target.
Usage
Execute this workflow when you need to extract structured data from a relational database into a Mage data pipeline. Common triggers include:
- You have a SQL database (PostgreSQL, MySQL, BigQuery, Snowflake, etc.) containing data needed for analytics or downstream processing
- You need incremental extraction that only fetches records modified since the last run
- You need automatic schema discovery to detect available tables and their column types
- You need paginated extraction for large tables that cannot fit in memory
Execution Steps
Step 1: Configuration and Connection Setup
Prepare the source connector configuration including database credentials and connection parameters. The configuration is loaded from a JSON file or settings dictionary containing host, port, database name, username, password, and optional parameters like SSH tunnel settings, SSL certificates, and schema filters.
Key considerations:
- Each database type has its own required configuration keys (see connector README files)
- SSH tunnel configuration is supported for PostgreSQL, MySQL, and other connectors
- Connection parameters are validated via the test_connection method before extraction begins
Step 2: Schema Discovery
Automatically discover available tables, columns, and their data types by querying the database's information_schema. The discovery process builds a Singer-compatible catalog containing stream definitions with JSON Schema types mapped from SQL column types.
What happens:
- The SQL base class queries information_schema.columns for all tables in the configured schema
- SQL data types are mapped to JSON Schema types (e.g., integer, varchar to string, json to object, timestamp to string with date-time format)
- Primary keys and unique constraints are extracted from COLUMN_KEY metadata
- Each table becomes a stream entry in the catalog with its schema, key properties, and replication method
- The catalog is output as JSON following the Singer specification
Step 3: Catalog Selection and Configuration
Select which streams (tables) and columns to extract from the discovered catalog. Each stream can be configured with a replication method, bookmark properties for incremental sync, and Mage-specific extensions like partition keys and unique conflict methods.
Key considerations:
- Streams must be explicitly marked as selected in the catalog metadata
- Columns are marked with inclusion levels: automatic (always extracted), available (user-selectable), or unsupported
- Replication method choices: FULL_TABLE (extract everything), INCREMENTAL (use bookmarks), LOG_BASED (CDC)
- Mage catalog extensions include auto_add_new_fields, bookmark_property_operators, and unique_conflict_method
Step 4: State Initialization
Load the sync state from a previous run to enable incremental extraction. The state contains bookmark values per stream that indicate where the last extraction ended, allowing the current run to pick up from that point.
What happens:
- State JSON is loaded from file if provided via the --state CLI argument
- State structure contains bookmarks per stream: each bookmark maps a column name to its last known value
- For first-time runs, state is empty and the full table is extracted
- For subsequent runs, bookmark values are used to build WHERE clause filters
Step 5: Paginated Data Extraction
Extract data from each selected stream in batches using SQL queries with LIMIT/OFFSET pagination. For incremental streams, a WHERE clause filters records based on bookmark values from the previous state.
What happens:
- For each selected stream, a SQL SELECT query is constructed with the selected columns
- If incremental, a WHERE clause is added: bookmark_column >= last_bookmark_value
- Results are ordered by bookmark columns first, then key columns, then unique constraint columns
- Data is fetched in configurable batch sizes (default BATCH_FETCH_LIMIT) using LIMIT/OFFSET
- Each batch of rows is converted from tuples to dictionaries keyed by column names
- The system pauses briefly between batches to avoid overwhelming the database
Step 6: Singer Message Emission
Emit Singer-protocol messages to stdout for each stream. This includes SCHEMA messages (stream metadata), RECORD messages (individual data rows), and STATE messages (bookmark checkpoints) that can be consumed by any Singer-compatible target.
What happens:
- Before records, a SCHEMA message is written containing the stream's JSON Schema, key properties, and bookmark properties
- Each extracted row is emitted as a RECORD message with the stream name and record data
- For incremental streams, STATE messages are emitted periodically with updated bookmark values
- After all records for a stream, a final STATE message captures the latest bookmark position
- The bookmark value tracks the maximum value seen for the replication key column
Step 7: State Persistence
Persist the final sync state so the next run can resume from where this extraction ended. The state captures bookmark values for each stream, enabling efficient incremental extraction across runs.
Key considerations:
- State is emitted as a STATE Singer message and captured by the downstream target
- The target writes state to a file that is passed back to the source on the next run
- State enables resume capability: the same extraction can run repeatedly without reprocessing old data
- For FULL_TABLE replication, no bookmarks are maintained (full re-extraction each run)