Streaming meets batch without tangling

Streaming and batch can coexist or compete. They coexist when you decide per dataset which track owns the write path. They compete when two teams both write to orders and disagree on which is authoritative. Draw the ownership lines first, wire the tools second.

Dataset ownership by track

Orders own the batch track. Sensor readings own the streaming track. They meet in the warehouse.

One owner per dataset. Two tracks converge in the warehouse.
kafka/producer.py
python
from confluent_kafka import Producer
import json

producer = Producer({'bootstrap.servers': 'kafka:29092'})


def publish(sensor_id: str, reading: dict) -> None:
    payload = json.dumps(reading).encode('utf-8')
    producer.produce(
        topic='sensor_readings',
        key=sensor_id.encode('utf-8'),
        value=payload,
        on_delivery=lambda err, msg: None if not err else print(f'delivery failed: {err}'),
    )
    producer.poll(0)

A minimal Kafka producer using confluent-kafka-python. Messages are JSON-encoded. The key ensures all readings for a single sensor hit the same partition, which preserves ordering per sensor.

Because downstream often wants ordering within a sensor but tolerates any order across sensors. Keying on sensor_id guarantees all readings for one device land on one partition, and Kafka preserves order within a partition. Random partitioning gives you more parallelism but breaks per-sensor ordering.

spark/spark_streaming_job.py
python
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window, avg
from pyspark.sql.types import StructType, StringType, DoubleType

spark = SparkSession.builder.appName('sensor-streaming').getOrCreate()

schema = StructType().add('sensor_id', StringType()).add('temp', DoubleType()).add('ts', StringType())

raw = (
    spark.readStream
    .format('kafka')
    .option('kafka.bootstrap.servers', 'kafka:29092')
    .option('subscribe', 'sensor_readings')
    .load()
)

parsed = raw.select(from_json(col('value').cast('string'), schema).alias('r')).select('r.*')

per_minute = (
    parsed.withColumn('ts', col('ts').cast('timestamp'))
          .groupBy(window(col('ts'), '1 minute'), col('sensor_id'))
          .agg(avg('temp').alias('avg_temp'))
)

(per_minute.writeStream
    .format('jdbc')
    .option('checkpointLocation', '/chk/sensor-aggregate')
    .outputMode('append')
    .trigger(processingTime='30 seconds')
    .start('agg_sensor_minute'))

The Spark structured streaming consumer reads from Kafka, aggregates per minute, and writes to Postgres. Structured streaming gives you exactly-once semantics if you enable checkpointing, which you always should.

Validation checklist: Verify the streaming track

Loading practice…

Quiz: Quiz

Loading practice…