Streaming Data Systems

Real-time data ingestion and stream processing with Apache Kafka, MQTT, and NATS JetStream. Covers producers, consumers, and stream processing patterns for data engineering pipelines.

Quick Comparison

Feature	Apache Kafka	MQTT	NATS JetStream
Use Case	High-throughput event streaming	IoT, mobile, constrained devices	Cloud-native, microservices
Throughput	Millions/sec	Thousands/sec	Hundreds of thousands/sec
Durability	Disk-based log, replayable	Ephemeral (configurable)	Disk-based persistence
Ordering	Per-partition	N/A (topic-based)	Per-subject
Python Client	confluent-kafka	paho-mqtt	nats-py
Best For	Event sourcing, CDC, log aggregation	Sensor data, telemetry	Service-to-service messaging

When to Use Which?

• Kafka: High-volume event streams, log aggregation, CDC, data lake ingestion
• MQTT: IoT devices, mobile push, constrained networks
• NATS JetStream: Microservices, request-reply, cloud-native, simpler ops than Kafka

Skill Dependencies

• @data-engineering-core - Process stream data with Polars/DuckDB
• @data-engineering-orchestration - Orchestrate stream processing jobs
• @data-engineering-quality - Validate streaming data
• @data-engineering-storage-lakehouse - Persist streams to Delta/Iceberg

---

Detailed Guides

Apache Kafka

Install: pip install confluent-kafka

Producer

from confluent_kafka import Producer
import socket
import json

def delivery_report(err, msg):
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Delivered to {msg.topic()} [{msg.partition()}]")

conf = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': socket.gethostname(),
    'acks': 'all'  # Wait for all replicas
}

producer = Producer(conf)

# Send messages asynchronously
for i in range(100):
    data = {'id': i, 'event': 'user_activity', 'value': i * 10}
    producer.produce(
        topic='user_activity_events',
        key=str(i),
        value=json.dumps(data).encode('utf-8'),
        callback=delivery_report
    )
    producer.poll(0)  # Trigger callbacks

producer.flush()  # Wait for delivery

# With Schema Registry (Avro)
from confluent_kafka.serialization import StringSerializer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer

schema_registry_client = SchemaRegistryClient({'url': 'http://localhost:8081'})
avro_serializer = AvroSerializer(schema_registry_client, schema_str)

producer = SerializingProducer({
    'bootstrap.servers': 'localhost:9092',
    'key.serializer': StringSerializer('utf_8'),
    'value.serializer': avro_serializer,
})

Consumer

from confluent_kafka import Consumer, KafkaError

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my_consumer_group',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,  # Manual commit
    'max.poll.interval.ms': 300000
}

consumer = Consumer(conf)
consumer.subscribe(['user_activity_events'])

try:
    while True:
        msg = consumer.poll(timeout=1.0)

        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            raise KafkaException(msg.error())

        # Process message
        data = json.loads(msg.value().decode('utf-8'))
        print(f"Received: {data}")

        # Manual commit after processing
        consumer.commit(asynchronous=False)

except KeyboardInterrupt:
    pass
finally:
    consumer.close()

Stream Processing (ksqlDB pattern)

For complex transformations, use ksqlDB or Kafka Streams. REST API example:

import requests

ksql_query = {
    "ksql": """
        SELECT id,
               COUNT(*) AS event_count,
               SUM(value) AS total_value
        FROM user_activity_events
        WINDOW TUMBLING (SIZE 1 MINUTE)
        GROUP BY id
        EMIT CHANGES
    """
}

response = requests.post("http://localhost:8088/query", json=ksql_query)
for line in response.iter_lines():
    print(line)

MQTT for IoT

Install: pip install paho-mqtt

Publisher

import paho.mqtt.client as mqtt
import json
import time

broker = "broker.emqx.io"
topic = "iot/sensors/temperature"

client = mqtt.Client(client_id="publisher_1", protocol=mqtt.MQTTv5)

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("Connected to broker")
    else:
        print(f"Connection failed: {rc}")

client.on_connect = on_connect
client.connect(broker)
client.loop_start()

for i in range(10):
    payload = {
        "sensor_id": "temp_sensor_1",
        "temperature": 20 + i * 0.5,
        "humidity": 45 + i,
        "timestamp": time.time()
    }
    client.publish(
        topic=topic,
        payload=json.dumps(payload),
        qos=1  # At-least-once delivery
    )
    time.sleep(5)

client.loop_stop()
client.disconnect()

Subscriber

def on_connect(client, userdata, flags, rc):
    client.subscribe(topic, qos=1)

def on_message(client, userdata, msg):
    payload = json.loads(msg.payload.decode("utf-8"))
    print(f"[{msg.topic}] {payload}")

client.on_connect = on_connect
client.on_message = on_message
client.connect(broker)
client.loop_forever()

NATS JetStream

Install: pip install nats-py

Producer/Consumer

import asyncio
import nats

async def main():
    nc = await nats.connect("nats://localhost:4222")
    js = nc.jetstream()

    # Create stream
    await js.add_stream(
        name="events",
        subjects=["events.*"],
        storage="file",
        max_msgs=10000,
        max_age=3600
    )

    # Publish
    await js.publish("events.page_loaded", b'{"page": "/home"}')

    # Push consumer
    sub = await js.subscribe("events.*", durable="worker-1")
    async for msg in sub:
        print(f"Received: {msg.data.decode()}")
        await msg.ack()

    await nc.close()

asyncio.run(main())

Work Queue Pattern

async def worker(name: str):
    nc = await nats.connect("nats://localhost:4222")
    js = nc.jetstream()
    sub = await js.subscribe("jobs.*", durable="workers", queue_group="processing")

    async for msg in sub:
        job_data = msg.data.decode()
        print(f"Worker {name} processing: {job_data}")
        await msg.ack()
    await nc.close()

---

Production Patterns

Idempotent Processing

Stream processors may receive duplicates. Design idempotent consumers:

processed_ids = load_checkpoint()  # From DB/Redis
if msg.id in processed_ids:
    ack()  # Skip duplicate
else:
    process(msg)
    save_checkpoint(msg.id)
    ack()

Batch Processing

# Accumulate messages before writing to reduce DB load
batch = []
while True:
    msg = consumer.poll(timeout=0.1)
    if msg:
        batch.append(msg.value())
    if len(batch) >= BATCH_SIZE or timeout_reached:
        write_to_db(batch)
        consumer.commit()  # Commit after batch write
        batch.clear()

Error Handling (Dead Letter Queue)

try:
    process(msg)
except Exception as e:
    if is_retryable(e):
        nack(requeue=True)  # Retry
    else:
        produce_to_dlq(msg, str(e))  # Send to dead letter queue
        ack()

Schema Evolution

• Use Avro/Protobuf with Schema Registry for compatibility
• Evolve schemas additively (new fields optional, old fields preserved)
• Register schemas per topic

---

References

• Apache Kafka Documentation
• confluent-kafka Python Client
• MQTT 5.0 Specification
• NATS JetStream Documentation
• @data-engineering-orchestration - Orchestrate consumers/producers as flows