Note: confluent-kafka 2.6.0 issues on Azure

See: confluentinc/confluent-kafka-python#1838

Add confluent-kafka == 2.5.3 to your dependencies list

Snapstream

Snapstream provides a data-flow model to simplify development of stateful streaming applications.

Installation

pip install snapstream

Usage

We snap iterables to user functions, and process them in parallel when we call stream:

We pass the callable print to print out the return value. Multiple iterables and sinks can be passed.

from snapstream import snap, stream

@snap(range(5), sink=[print])
def handler(msg):
    yield f'Hello {msg}'

stream()

Hello 0
Hello 1
Hello 2
Hello 3
Hello 4

To try it out for yourself, spin up a local kafka broker with docker-compose.yml, using localhost:29091 to connect:

docker compose up broker -d

Use the cli tool to inspect Topic/Cache:

snapstream topic emoji --offset -2

>>> timestamp: 2023-04-28T17:31:51.775000+00:00
>>> offset: 0
>>> key:
🏆

Features

• snapstream.snap: bind streams (iterables) and sinks (callables) to user defined handler functions
• snapstream.stream: start streaming
• snapstream.Topic: consume from (iterable), and produce to (callable) kafka using confluent-kafka
• snapstream.Cache: store data to disk using rocksdict
• snapstream.Conf: set global kafka configuration (can be overridden per topic)
• snapstream.codecs.AvroCodec: serialize and deserialize avro messages
• snapstream.codecs.JsonCodec: serialize and deserialize json messages