Stream Maps Syntax & Functions

Relevant source files

The following files were used as context for generating this wiki page:

• cookiecutter/mapper-template/{{cookiecutter.mapper_id}}/AGENTS.md
• singer_sdk/mapper.py
• singer_sdk/mapper_base.py
• packages/meltano-mapper-custom/mapper_custom/mapper.py
• tests/core/test_mapper.py

Stream Maps provide a powerful, inline transformation layer within the Meltano Singer SDK. They allow developers and users to transform records in-flight as data moves from a source (Tap) to a destination (Target). This capability includes renaming fields, calculating new values, filtering records, aliasing streams, and masking sensitive information.

The system operates by evaluating Python-like expressions against each record. These expressions have access to the record's properties, configuration settings, and a suite of built-in functions. By defining these maps in the plugin configuration, users can implement complex data engineering logic without writing custom standalone scripts.

Sources: cookiecutter/mapper-template/{{cookiecutter.mapper_id}}/AGENTS.md, singer_sdk/mapper.py:270-271

Architecture & Components

The Stream Maps system is built around the PluginMapper class, which manages multiple StreamMap instances. Each StreamMap is responsible for the transformation logic of a specific stream alias.

Core Classes

• InlineMapper: An abstract base class that implements both BaseSingerReader and BaseSingerWriter, allowing it to sit between Singer components and process messages in real-time.
• PluginMapper: The primary coordinator that registers stream schemas and matches incoming streams to their configured transformations.
• CustomStreamMap: Handles the evaluation of user-defined expressions using a specialized evaluator.
• RemoveRecordTransform: A specialized map used to drop records or entire streams.

Sources: singer_sdk/mapper_base.py:26-31, singer_sdk/mapper.py:612-620, packages/meltano-mapper-custom/mapper_custom/mapper.py:59-65

Data Flow Diagram

The following diagram illustrates how messages flow through an InlineMapper using Stream Maps.

The InlineMapper intercepts standard Singer messages, applying transformations to RECORD and SCHEMA messages based on the stream_maps configuration while generally passing STATE messages through to preserve bookmarks.

Sources: singer_sdk/mapper_base.py:51-70, packages/meltano-mapper-custom/mapper_custom/mapper.py:89-140

Expression Syntax

Stream maps use a Python-based expression syntax evaluated via the simpleeval library. These expressions can reference several internal variables and operators.

Reserved Operators

Operator	Description
__filter__	A boolean expression to determine if a record should be included.
__source__	Specifies a different source stream for a new aliased stream.
__else__	Determines the behavior for fields or streams not explicitly mapped (e.g., null to exclude).
__alias__	Renames an entire stream.
__key_properties__	Overrides the primary keys of the transformed stream.

Sources: singer_sdk/mapper.py:44-50, cookiecutter/mapper-template/{{cookiecutter.mapper_id}}/AGENTS.md:52-57

Context Variables

When an expression is evaluated, the following names are available in the namespace:

• _ or record: The full original record dictionary.
• self: The original value of the property being transformed (only available within field-level maps).
• config: The stream_map_config dictionary provided in the plugin settings.
• __stream_name__: The alias of the current stream.
• __original_stream_name__: The name of the source stream before any aliasing.
• fake: A Faker instance (if the faker extra is installed and configured).

Sources: singer_sdk/mapper.py:348-360, tests/core/test_mapper.py:595-603

Built-in Functions

The SDK provides several utility functions and modules available directly within mapping expressions.

Transformation Functions

• md5(string): Returns the MD5 hex digest of the input string.
• sha256(string): Returns the SHA256 hex digest of the input string.
• bool(value): Standard Python boolean conversion.
• int(value): Standard Python integer conversion.
• str(value): Standard Python string conversion.
• float(value): Converts value to a float (maps to DecimalType in the output schema).

Sources: singer_sdk/mapper.py:55-75, singer_sdk/mapper.py:328-335, tests/core/test_mapper.py:127-135

Modules

• datetime: Access to the Python datetime module (e.g., datetime.date.today()).
• json: Access to the json module, commonly used for json.dumps(record).

Sources: singer_sdk/mapper.py:329-331, tests/core/test_mapper.py:682-687

Stream Configuration Examples

Record Filtering

Filtering is achieved using the __filter__ key. Records are excluded if the expression evaluates to False.

stream_maps = {
    "transactions": {
        "__filter__": "amount > 0 and status == 'completed'"
    }
}

Sources: cookiecutter/mapper-template/{{cookiecutter.mapper_id}}/AGENTS.md:104-110, singer_sdk/mapper.py:442-450

Stream Aliasing and Cloning

Users can rename streams or create multiple output streams from a single source.

stream_maps = {
    "customers": {
        "__alias__": "users"
    },
    "customers_v2": {
        "__source__": "customers",
        "new_field": "'static_value'"
    }
}

Sources: singer_sdk/mapper.py:730-745, cookiecutter/mapper-template/{{cookiecutter.mapper_id}}/AGENTS.md:144-158

Property Mapping & Masking

Fields can be renamed by mapping the new key to the old key, or values can be computed.

stream_maps = {
    "users": {
        "user_id": "id",
        "email_hash": "md5(email)",
        "full_name": "first_name + ' ' + last_name",
        "password": None  # Exclude property
    }
}

Sources: cookiecutter/mapper-template/{{cookiecutter.mapper_id}}/AGENTS.md:72-82, tests/core/test_mapper.py:140-149

Implementation Logic

The CustomStreamMap class performs a "Dry Run" type evaluation to determine the resulting schema when a SCHEMA message is received.

Schema Inference Logic

The system attempts to infer the JSON Schema type based on the expression string:

• Expressions wrapped in int() map to integer.
• Expressions wrapped in float() map to number.
• Expressions starting with datetime.date map to date.
• Expressions starting with datetime.datetime map to date-time.
• Most other expressions default to string if they involve string literals or unknown function calls.

Sources: singer_sdk/mapper.py:382-414

Class Relationship Diagram

This hierarchy shows how PluginMapper aggregates various StreamMap implementations to provide transformation services to the InlineMapper.

Sources: singer_sdk/mapper_base.py:26-31, singer_sdk/mapper.py:77-183, singer_sdk/mapper.py:612-615

Summary

Stream Maps Syntax & Functions provide a high-level abstraction for data transformation in Singer pipelines. By utilizing the simpleeval engine, the SDK allows for complex logic—including hashing, filtering, and aliasing—to be defined entirely within configuration. This ensures that mappers remain stateless and lightweight while providing significant flexibility for data normalization and privacy compliance.

Sources: cookiecutter/mapper-template/{{cookiecutter.mapper_id}}/AGENTS.md:213-220, singer_sdk/mapper.py:28-33