Record-based Sinks
Relevant source files
The following files were used as context for generating this wiki page:
Record-based sinks are a core component of the Meltano Singer SDK's target implementation, responsible for receiving, validating, and loading data into a destination. Within the SDK's architecture, a Target object orchestrates one or more Sink objects, which correspond to specific data streams. Record-based sinks specifically refer to implementations where data is processed either as individual records or accumulated in batches for bulk loading.
These sinks provide a standardized interface for handling the Singer protocol's lifecycle, including schema initialization, record transformation, validation against JSON schemas, and state management. Depending on the destination's requirements, developers can choose between "Per record" serialization for streaming/API-based loads or "Per batch" and SQL Serialization for high-throughput database operations.
Sources: singer_sdk/target_base.py:27-35, cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md:38-40
Architecture and Data Flow
The Sink class acts as the destination-specific handler for a single stream. When a Target receives a RECORD message, it routes the data to the appropriate Sink based on the stream name. If the sink does not yet exist, the target initializes it using the most recent SCHEMA message received.
This diagram shows the high-level flow of a record through the target and into the sink's processing logic.
Sources: singer_sdk/target_base.py:252-290, singer_sdk/sinks/core.py:355-380
Key Components
- Target Class: The main entry point that manages configuration and orchestrates sinks. It handles incoming Singer messages (
RECORD,SCHEMA,STATE) and routes them to the correct sink instance. - Sink Class: The abstract base class (
singer_sdk.sinks.Sink) that developers inherit from to implement destination-specific loading logic. - JSONSchemaValidator: A utility used by sinks to ensure incoming records conform to the defined schema before they are processed.
Sources: singer_sdk/target_base.py:46-52, singer_sdk/sinks/core.py:116-121, CLAUDE.md:85-88
Processing Modes
The SDK supports different processing modes tailored to different destination types. These are primarily defined by how the process_record and process_batch methods are implemented.
Per Record Serialization
In this mode, process_record() is used to immediately send data to the destination. This is ideal for REST APIs, message queues, or webhooks where real-time delivery is preferred over high-volume throughput. It typically has a lower memory footprint because records are not held in memory.
def process_record(self, record: dict, context: dict) -> None:
"""Process a single record immediately."""
# Example: Send to an API
self.session.post(self.api_url, json=record)
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md:46-60
Per Batch Serialization
This mode accumulates records in a context object (typically a list in records_to_drain) until a maximum size is reached. Once the sink is "full," process_batch() is triggered to perform a bulk operation. This is more efficient for file uploads or bulk APIs.
Key properties for batching:
max_size: The threshold for records per batch (default is 10,000).batch_size_rows: A configuration option to override the default batch size.
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md:62-79, singer_sdk/sinks/core.py:161-161, singer_sdk/sinks/core.py:315-325
SQL Serialization
For database destinations, the SDK provides a specialized SQLSink. This sink leverages a SQLConnector to handle DDL operations (like creating tables) and supports optimized bulk loading methods.
The class hierarchy shows how SQLSink extends the base sink functionality with database-specific capabilities.
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/{{cookiecutter.library_name}}/sinks_sql.py:22-55, cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md:81-100
Validation and Metadata
Sinks automatically handle several data integrity and bookkeeping tasks before the data reaches the process_record method.
| Feature | Description | Source |
|---|---|---|
| JSON Schema Validation | Validates records against the stream's schema; can be configured to fail on errors. | sinks/core.py:133-156 |
| SDC Metadata | Optionally injects _sdc columns (e.g., _sdc_extracted_at, _sdc_received_at) for auditability. | sinks/core.py:397-420 |
| Timestamp Parsing | Automatically converts ISO-8601 strings into Python datetime objects. | sinks/core.py:537-570 |
| Primary Key Check | Ensures records contain all defined key_properties. | sinks/core.py:521-535 |
Record Lifecycle Sequence
Configuration Options
Sinks inherit configuration from the parent Target. Common SDK-level settings affect sink behavior:
| Config Option | Type | Default | Description |
|---|---|---|---|
add_record_metadata | Boolean | False | Whether to add _sdc metadata columns to the records. |
validate_records | Boolean | True | Enables or disables JSON schema validation for every record. |
batch_size_rows | Integer | 10000 | The maximum number of rows to accumulate before a batch drain. |
load_method | String | append | Defines how data is loaded (e.g., append vs. overwrite). |
Sources: singer_sdk/target_base.py:463-475, singer_sdk/sinks/core.py:246-252, singer_sdk/sinks/core.py:325-333
Conclusion
Record-based sinks provide a robust framework for implementing Singer targets. By abstracting the complexities of the Singer protocol—such as schema evolution, message parsing, and batching—they allow developers to focus on the specific logic required to write data to their chosen destination. Whether using simple per-record processing or complex SQL bulk inserts, these sinks ensure that data remains consistent and compliant with the Singer specification.