Batch-based Sinks
Relevant source files
The following files were used as context for generating this wiki page:
Batch-based Sinks are a specialized component of the Meltano Singer SDK designed to handle data loading in groups rather than on a per-record basis. This architecture is optimized for high-throughput destinations such as bulk APIs, file uploads, and data warehouses where processing individual records would be inefficient.
The primary purpose of a batch sink is to accumulate records in memory (up to a defined threshold) and then execute a single processing operation on the entire collection. This approach significantly reduces the overhead associated with network requests and transaction management in the target destination.
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md, CLAUDE.md
Architecture and Data Flow
The batch sink operates as a middle layer between the incoming Singer record stream and the final destination. It inherits from the base Sink class but overrides the record-handling logic to implement a buffering mechanism.
Key Components
- BatchSink Class: The base class for all batch-oriented loaders. It manages the accumulation of records and triggers the processing logic when limits are reached.
- Context Object: A dictionary-like structure used to store the current batch of records and any metadata required for the processing cycle.
- Max Size Threshold: A configuration parameter that determines how many records can be held in memory before a batch must be flushed.
Batch Processing Workflow
The following diagram illustrates how records flow through a Batch Sink, from initial receipt to final processing at the destination.
The flow shows the conditional logic where records are either buffered or trigger a batch execution based on the max_size configuration.
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md
Implementation Details
Developers implementing a Batch Sink primarily interact with the process_batch method. Unlike process_record (used in streaming sinks), process_batch is only called once per group of records.
Core Methods
| Method | Description |
|---|---|
process_record(record, context) | Overridden to append the record to the context instead of immediate processing. |
process_batch(context) | The primary implementation point where the list of records is sent to the destination. |
prepare_record(record) | Used to transform or format individual records before they are added to the batch. |
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md
Example Implementation
def process_batch(self, context: dict) -> None:
"""Process a batch of records."""
records = context["records"]
# Logic to transform the entire batch (e.g., to CSV or JSONL)
# Upload to destination (e.g., S3, BigQuery, or a Bulk API)
# Handle partial failures if the API supports it
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md
Configuration and Constraints
Batch sinks are governed by specific configuration properties and performance considerations to ensure system stability.
Performance Properties
- max_size: The maximum number of records allowed in a single batch. The default is typically 10,000 records.
- Memory Management: Because records are stored in memory, large batch sizes can lead to OOM (Out of Memory) errors. It is recommended to start with 1,000–5,000 records and adjust based on record size.
- Batch Encoders: The SDK provides built-in encoders for common batch formats, such as
jsonlandparquet.
Encoders and Plugins
The SDK supports several batch encoding formats via entry points, allowing for efficient serialization of data before transfer.
| Encoder | Description |
|---|---|
jsonl | Newline-delimited JSON, suitable for many bulk APIs and cloud storage. |
parquet | Columnar storage format, highly optimized for analytical workloads. |
Sources: pyproject.toml, cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md
Error Handling and State
Batch sinks must handle errors at the batch level. If a process_batch call fails, the implementation must decide whether to raise a RetriableAPIError for the SDK to attempt the entire batch again or a FatalAPIError to stop the sync.
State messages are received and forwarded by the sink. In a batch context, state is typically only emitted once a batch has been successfully acknowledged by the destination, ensuring that the tap does not advance its bookmarks for data that hasn't been safely loaded.
Sources: CLAUDE.md, cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md
Conclusion
Batch-based Sinks provide a robust framework for high-volume data loading within the Meltano Singer SDK. By balancing memory usage with throughput through the max_size threshold and centralized process_batch logic, they enable efficient integration with modern data stack destinations. Implementation requires careful consideration of memory footprints and robust error handling to maintain data integrity during bulk operations.