Enrich a stream with reference data by key lookup in Python

Uses streamz to join each incoming record to a reference dictionary by name, adding department and level fields or defaults.

Medium Python 3.9+ Aug 9, 2026 Data pipelines & processing 14 views 0 copies

streamz streaming join enrichment reference-data

Requires third-party packages — install first

pip install streamz

Python code

21 lines

Python 3.9+

from streamz import Stream

reference = {"alice": {"dept": "eng", "level": 3}, "bob": {"dept": "sales", "level": 5}}

def enrich(record):
    name = record.get("name")
    ref = reference.get(name)
    joined = dict(record)
    if ref:
        joined.update(ref)
    else:
        joined["dept"] = "unknown"
        joined["level"] = 0
    return joined

source = Stream()
source.map(enrich).sink(print)

if __name__ == "__main__":
    source.emit({"name": "alice", "score": 42})
    source.emit({"name": "carol", "score": 17})

Output

stdout

{'name': 'alice', 'score': 42, 'dept': 'eng', 'level': 3}
{'name': 'carol', 'score': 17, 'dept': 'unknown', 'level': 0}

How it works

The Stream object acts as a pipeline where each emitted value flows through map. The enrich function performs a dictionary lookup on the reference dict by the record's name key. If found, it merges the reference fields into a copy of the record; otherwise it supplies default department and level values. .sink(print) sends the enriched records to stdout. This pattern mirrors stream–table joins common in streaming data pipelines.

Common mistakes

Mutating the original record in-place instead of copying with dict(record), corrupting upstream state
Assuming the lookup key exists without using .get() to avoid KeyError
Using the same reference dict across threads without synchronization in concurrent streams

Variations

Use a defaultdict returning default values for missing keys instead of an if/else
Apply the mapping with operator.itemgetter for simple single-field enrichment

Real-world use cases

Joining user session events to static user profile dimensions in a real-time analytics pipeline.
Enriching clickstream records with device metadata from a nightly-loaded reference table.
Augmenting order events with warehouse zone and carrier data before downstream aggregation.

Enrich a stream with reference data by key lookup in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Data pipelines & processing

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Data pipelines & processing

Keep learning

Tutorials

Quizzes