Core Data Streams
Relevant source files
The following files were used as context for generating this wiki page:
Core Data Streams represent the foundational abstraction for extracting data from various sources within the Meltano Singer SDK. These streams are responsible for defining the structure (schema), replication logic, and retrieval mechanisms for individual data endpoints such as users, orders, or transactions. By adhering to the Singer specification, core streams ensure that data extractors (taps) provide consistent, high-quality data to downstream loaders (targets).
The primary component is the Stream abstract base class, which manages stream-level metadata, state tracking, and record generation. It provides built-in support for multiple replication methods, including full table syncs and incremental syncing using bookmarks.
Sources: singer_sdk/streams/core.py:38-66, cookiecutter/tap-template/{{cookiecutter.tap_id}}/AGENTS.md:27-35
Stream Architecture and Lifecycle
The lifecycle of a data stream involves initialization with a tap instance and schema, followed by a synchronization process that manages state, schema emission, and record extraction.
Core Components
| Component | Description |
|---|---|
Stream | The abstract base class providing the framework for data extraction. |
StreamStateManager | Manages replication keys, bookmarks, and partition-level state. |
CatalogEntry | Represents the stream's metadata and schema within a Singer catalog. |
StreamMap | Handles in-flight transformations and filtering of records. |
Sources: singer_sdk/streams/core.py:100-140, singer_sdk/singerlib/catalog.py:236-255
Synchronization Flow
The synchronization flow ensures that the target system receives the stream's schema before any data records. It also manages state signposts to prevent data loss during incremental syncs.
This diagram illustrates the sequence of operations during a stream's sync() execution.
Sources: singer_sdk/streams/core.py:771-828
Data Replication and State Management
Streams support three primary replication methods to determine how data is extracted over time.
Replication Methods
- FULL_TABLE: Extracts all records from the source every time.
- INCREMENTAL: Uses a
replication_key(e.g.,updated_at) to extract only records changed since the last sync. - LOG_BASED: Relies on database logs for change data capture.
Sources: singer_sdk/singerlib/catalog.py:27-30, singer_sdk/streams/core.py:539-545
State Tracking logic
State is managed via the StreamStateManager, which tracks bookmarks for individual streams or partitions. For incremental streams, the SDK seeds the sync using get_starting_replication_key_value() to retrieve the last successfully processed bookmark from the provided state.
The sequence shows how a stream interacts with its state manager to resume incremental syncs and update bookmarks.
Sources: singer_sdk/streams/core.py:235-260, singer_sdk/streams/core.py:651-678
Stream Relationships and Partitioning
The SDK allows for complex data structures through parent-child stream relationships and partitioning.
Parent-Child Streams
Child streams can depend on the context provided by a parent stream. For example, an EpicIssuesStream may require a group_id and epic_id from a parent EpicsStream.
parent_stream_type: Defines the class of the parent stream.get_child_context(): Extracts necessary identifiers from a parent record to pass to the child stream's sync process.
Sources: packages/meltano-tap-gitlab/tap_gitlab/streams.py:168-185, singer_sdk/streams/core.py:92-96
Partitioning
Partitioning allows a single stream to be split into multiple logical units of work, each with its own state bookmark. This is common when syncing data across multiple project IDs or account IDs.
class ProjectBasedStream(GitlabStream):
@property
def partitions(self) -> list[dict]:
return [{"project_id": pid} for pid in self.config["project_ids"]]
Sources: packages/meltano-tap-gitlab/tap_gitlab/streams.py:73-81, singer_sdk/streams/core.py:619-630
Schema and Type Conformance
Core Data Streams enforce data quality through JSON Schema validation and type conformance.
Conformance Levels
The TYPE_CONFORMANCE_LEVEL attribute determines how strictly the SDK converts record data to match the schema:
RECURSIVE: Converts types through all nested levels (Default).ROOT_ONLY: Only converts top-level fields.NONE: No type conversion performed.
Sources: singer_sdk/streams/core.py:53-73
Metadata Mapping
Streams use a MetadataMapping to manage properties like selection, inclusion (available, automatic, unsupported), and forced replication methods.
| Metadata Field | Type | Description |
|---|---|---|
inclusion | Enum | Determines if a property is always included or optional. |
selected | Boolean | Indicates if the user has selected this property for sync. |
replication-key | String | Identifies the field used for incremental bookmarks. |
Sources: singer_sdk/singerlib/catalog.py:67-90, singer_sdk/singerlib/catalog.py:112-119
Batch Processing
For high-volume streams, the SDK supports batching records into files (e.g., JSONL or Parquet) instead of emitting individual RECORD messages. This is controlled via the BatchConfig.
get_batches(): Iterates through records and uses aBatcherto group them into files._write_batch_message(): Emits a BATCH message containing the file manifest and encoding details.
Sources: singer_sdk/streams/core.py:1012-1033, singer_sdk/streams/core.py:721-736
Summary
Core Data Streams provide the essential logic for robust data extraction. By combining automated state management, flexible parent-child relationships, and strict schema adherence, they enable developers to build maintainable Singer taps. The lifecycle ensures that every sync is identifiable by a clear schema and can be resumed incrementally through the standardized use of replication keys and state bookmarks.