Mapper Fundamentals
Relevant source files
The following files were used as context for generating this wiki page:
Mappers are specialized Singer plugins within the Meltano SDK designed to perform inline stream transformations. Positioned between a Tap (extractor) and a Target (loader), a Mapper modifies data in-flight, allowing for schema evolution, data masking, filtering, and field computation without requiring intermediate storage.
The architecture relies on the InlineMapper base class, which inherits from both BaseSingerReader and BaseSingerWriter. This dual nature enables the plugin to consume Singer messages from STDIN (or a file), apply transformation logic defined in stream_maps, and output modified Singer messages to STDOUT.
Sources: cookiecutter/mapper-template/{{cookiecutter.mapper_id}}/AGENTS.md:10-18, singer_sdk/mapper_base.py:27-28, README.md:46-52
Core Architecture
The Mapper system is built on an abstract foundation that handles the Singer message stream. The InlineMapper class serves as the primary interface for developers to implement custom transformation logic.
Message Processing Flow
The Mapper listens for incoming Singer messages and routes them to specific mapping methods based on the message type (SCHEMA, RECORD, STATE, etc.).
The diagram above illustrates how the InlineMapper intercepts the stream to apply transformations before re-emitting messages.
Sources: singer_sdk/mapper_base.py:53-75, packages/meltano-mapper-custom/mapper_custom/mapper.py:89-158
Key Classes and Capabilities
Mappers in the SDK utilize the following components:
| Component | Description |
|---|---|
InlineMapper | The abstract base class providing the plugin loop and CLI interface. |
PluginMapper | A helper class that parses stream_maps configuration and executes transformations. |
stream_maps | A configuration object defining the transformation rules for specific streams. |
Capabilities | Mappers natively support ABOUT, STREAM_MAPS, FLATTENING, and STRUCTURED_LOGGING. |
Sources: singer_sdk/mapper_base.py:30-36, packages/meltano-mapper-custom/mapper_custom/mapper.py:76-78, CLAUDE.md:82-89
Stream Maps Logic
Stream maps provide a declarative way to define how records and schemas are altered. They support Python-like expressions for dynamic computation.
Transformation Types
- Property Mapping: Renaming or transforming individual fields (e.g.,
{"new_name": "old_name"}). - Filtering: Using the
__filter__key to exclude records based on boolean logic (e.g.,"__filter__": "price > 100"). - Aliasing & Cloning: Renaming entire streams or creating multiple output streams from one input source using
__source__. - Exclusion: Removing fields by setting them to
nullor using__else__: nullto drop unmapped fields.
Sources: cookiecutter/mapper-template/{{cookiecutter.mapper_id}}/AGENTS.md:27-105
Expression Syntax
Mappers use simpleeval to evaluate string expressions. Supported operations include:
- Arithmetic:
price * quantity - String Formatting:
first_name + ' ' + last_name - Hashing:
md5(email)for PII masking. - Date Manipulation:
today()andstrftime()functions.
Sources: cookiecutter/mapper-template/{{cookiecutter.mapper_id}}/AGENTS.md:54-80, pyproject.toml:57
Implementation Details
Implementing map_record_message
The core logic for record transformation usually involves iterating through the configured stream maps and applying the transform method.
@override
def map_record_message(self, message_dict: dict) -> t.Generator[singer.Message, None, None]:
stream_id: str = message_dict["stream"]
time_extracted = utc_now()
for stream_map in self.mapper.stream_maps[stream_id]:
mapped_record = stream_map.transform(message_dict["record"])
if mapped_record is not None:
yield singer.RecordMessage(
stream=stream_map.stream_alias,
record=mapped_record,
version=message_dict.get("version"),
time_extracted=time_extracted,
)
Sources: packages/meltano-mapper-custom/mapper_custom/mapper.py:126-143
Configuration Schema
Mappers require a stream_maps object in their configuration. This is typically defined using the singer_sdk.typing module to ensure validation against the Mapper's settings.
config_jsonschema = th.PropertiesList(
th.Property(
"stream_maps",
th.ObjectType(
additional_properties=th.CustomType({
"type": ["object", "string", "null"],
"properties": {
"__filter__": {"type": ["string", "null"]},
"__source__": {"type": ["string", "null"]},
"__else__": {"type": ["null"]},
},
}),
),
required=True,
),
).to_dict()
Sources: packages/meltano-mapper-custom/mapper_custom/mapper.py:44-67
Conclusion
Mapper Fundamentals center on the ability to provide high-performance, stateless transformations within a Singer pipeline. By leveraging the InlineMapper base class and the stream_maps configuration, developers can implement complex data engineering tasks like PII masking, schema normalization, and stream filtering with minimal code. The system's integration with the Meltano SDK ensures that these mappers remain compliant with the Singer specification while offering flexible, expression-based logic.