Learn the core concepts, not deep administration. You need to understand how streaming data moves through Kafka well enough to build and reason about a pipeline — not to operate a production cluster.
Batch processing (Week 4) waits and processes data in scheduled chunks. Streaming processes each event as it arrives, continuously.
Why not just check the database more often? A natural first idea is "run the batch job every minute instead of every day." That breaks down for two reasons. First, checking a database every few seconds puts real, repeated load on it, just to ask "did anything change?" Second, checking on a timer always has a built-in delay — you're only ever as fast as your check interval. Kafka instead pushes events out the instant they happen. It also fully separates producers from consumers: an order-placing service doesn't need to know or care whether zero, one, or ten different systems are reading its events. It just publishes to a topic and moves on.
from kafka import KafkaProducer
import json
producer = KafkaProducer(
bootstrap_servers="localhost:9092",
value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)
producer.send("orders", {"order_id": 1042, "customer_id": 7, "total_amount": 59.99})
producer.flush()customer_id) always land on the same partition, preserving order for that key.flush() ensures buffered messages are actually sent before your script exits — a common gotcha when testing.Why keying matters, concretely — sending without a key lets Kafka spread messages round-robin across partitions for maximum throughput, but that scrambles per-entity ordering:
# no key: order_id 1042's status updates could land on different partitions,
# and partitions are consumed independently — a consumer might process
# "shipped" before "paid" simply because they took different paths
producer.send("order_events", {"order_id": 1042, "status": "paid"})
producer.send("order_events", {"order_id": 1042, "status": "shipped"})
# keyed by order_id: EVERY event for order 1042 always goes to the same partition,
# guaranteeing "paid" is read before "shipped" by whichever consumer processes it
producer.send("order_events", key=str(1042).encode(), value={"order_id": 1042, "status": "paid"})
producer.send("order_events", key=str(1042).encode(), value={"order_id": 1042, "status": "shipped"})Delivery guarantees are another setting worth knowing by name. acks=0 means "fire and forget" — fastest, but messages can silently get lost. acks=1 waits for the partition leader to confirm — this is the default, a reasonable middle ground. acks='all' waits for every in-sync replica to confirm — the slowest option, but a message only counts as sent once it's safely copied in multiple places, not just accepted by one server that could crash a second later.
from kafka import KafkaConsumer
import json
consumer = KafkaConsumer(
"orders",
bootstrap_servers="localhost:9092",
group_id="orders-processing-group",
value_deserializer=lambda v: json.loads(v.decode("utf-8")),
auto_offset_reset="earliest",
)
for message in consumer:
order = message.value
print(f"Processing order {order['order_id']}")auto_offset_reset="earliest" vs "latest" controls whether a new consumer starts from the beginning of the topic or only sees new messages.Committing offsets by hand is often a better idea. The example at the top auto-commits by default, which can quietly lose messages if your process crashes right after reading, but before it's actually finished the work. For anything writing to a database, only committing after a successful write is the safer, idempotency-friendly approach (same principle as Week 4):
consumer = KafkaConsumer(
"orders",
bootstrap_servers="localhost:9092",
group_id="orders-processing-group",
value_deserializer=lambda v: json.loads(v.decode("utf-8")),
enable_auto_commit=False, # take manual control of when an offset counts as "processed"
)
for message in consumer:
order = message.value
try:
write_to_postgres(order) # do the actual work first
consumer.commit() # only THEN mark this offset as processed
except Exception:
logger.exception("Failed to process order %s, will retry on next poll", order["order_id"])
# not committing means this message will be redelivered — the consumer just re-reads itThis is Kafka's version of "at-least-once" delivery — a crash between the write and the commit just means the message gets processed again. That's exactly why the downstream write (write_to_postgres) needs to be idempotent, same as any Airflow task.
A named stream of events — the Kafka equivalent of a table or a channel (e.g. orders, page_views, payment_events).
Retention, not permanence — unlike a database table, a Kafka topic doesn't keep data forever by default. Each topic has a retention policy, most commonly time-based:
# create a topic with a 7-day retention window
kafka-topics.sh --create --topic orders \
--partitions 3 --replication-factor 3 \
--config retention.ms=604800000Messages older than the retention window get deleted automatically, whether or not a consumer ever read them. That's why "replaying old events" only works within limits — a consumer group that's been offline for 8 days on a topic with 7-day retention has permanently lost that first day's events. Picking a retention period is a real decision: it's a trade-off between storage cost and how much history a consumer might genuinely need to replay.
Schema evolution is another thing worth planning for. Once producers and consumers are live, changing a message's shape — renaming a field, changing its type — can break every single consumer at once. Real Kafka setups usually use a schema registry to enforce rules about what changes are safe: adding a new field is fine, but removing or retyping an existing one needs a coordinated rollout across every producer and consumer.
customer_id should hit the same partition if order matters for that customer).Watching key-based partitioning happen — this is the mechanism behind "same key, same partition" from the Producer section, made concrete:
# conceptually, Kafka's default partitioner does roughly:
partition = hash(key) % number_of_partitions
# so every message keyed by customer_id=101 always hashes to the SAME partition number,
# for the life of the topic (as long as partition count doesn't change) —
# this is what guarantees per-customer ordering without needing a global orderThe catch: you can only add partitions, never remove them. And adding partitions changes which keys map to which partition (since the math above depends on the total count). That means adding partitions to an existing topic can briefly break ordering for any keys that get reassigned. It's worth choosing a partition count thoughtfully up front — usually based on how much consumer parallelism you'll want — rather than treating it as something to casually change later.
A sequential ID for each message within a partition — like a row number that only ever increases.
Rewinding on purpose — the "backfill" use case mentioned above, made concrete. Say a bug in a consumer mis-processed the last day's worth of events; once fixed, replay just that window:
consumer = KafkaConsumer(
"orders",
bootstrap_servers="localhost:9092",
group_id="orders-processing-group",
enable_auto_commit=False,
)
consumer.poll(timeout_ms=1000) # trigger partition assignment before seeking
for tp in consumer.assignment():
consumer.seek_to_beginning(tp) # or consumer.seek(tp, specific_offset) for a precise rewind
for message in consumer:
reprocess(message.value) # now re-reads from the beginning with the fixed logicThis is the streaming equivalent of Airflow's backfill (Week 7-8) — same core idea, fix the bug then re-run history against it — just triggered by manually resetting a consumer's position instead of picking a date range.
The two independent groups pattern, worth seeing explicitly since it's a common source of confusion — one topic, two totally separate consumer groups, each getting the full stream:
# consumer group 1: loads every order into the warehouse
KafkaConsumer("orders", group_id="warehouse-loader", ...)
# consumer group 2: powers a live "orders today" dashboard counter
KafkaConsumer("orders", group_id="live-dashboard", ...)
# both groups read ALL messages on the "orders" topic, completely independently —
# group_id is what defines a group's boundary; group 1's progress has zero
# effect on group 2's progress, each tracks its own offsets per partitionThis is exactly the fan-out pattern from the capstone diagrams elsewhere on this site: one event stream, several independent consumers downstream — a warehouse loader, a real-time dashboard, maybe a fraud-detection service — each reading through the same topic at its own pace, without slowing down or blocking any of the others.
In-sync replicas (ISR), and why acks='all' is actually safe: a replica only counts as "in sync" if it's genuinely caught up with the leader's latest messages. A replica that's lagging or disconnected automatically drops out of that set. This connects directly back to acks='all' from earlier — it really means "wait until every currently in-sync replica has the message," not "wait for literally every replica that was ever created." That way, one slow replica doesn't bring every write to a halt.
# min.insync.replicas: the other half of durability config, set per topic
kafka-topics.sh --alter --topic orders --config min.insync.replicas=2With replication.factor=3 and min.insync.replicas=2, a write using acks='all' only succeeds once at least 2 of the 3 replicas confirm it. That means Kafka can survive one broker going down without losing any acknowledged data, and without writes grinding to a halt. Together, these three settings — replication factor, min in-sync replicas, and producer acks — are what actually guarantee durability in a real Kafka setup. Worth understanding, even if you won't configure it by hand this early.
Build: Application → Kafka → Spark → PostgreSQL
# Spark Structured Streaming reading from Kafka
df = (
spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("subscribe", "orders")
.load()
)
orders = df.selectExpr("CAST(value AS STRING) as json") \
.select(from_json("json", order_schema).alias("data")) \
.select("data.*")
query = (
orders.writeStream
.format("jdbc") # or foreachBatch to upsert into Postgres
.start()
)Let's fill in the missing pieces: a minimal producer to generate test data, and the foreachBatch sink that actually upserts into Postgres. (The "jdbc" streaming format mentioned above can't do upserts on its own, so foreachBatch is the pattern people actually use in practice.)
# producer.py — simulates order events landing in Kafka
import json, time, random
from kafka import KafkaProducer
producer = KafkaProducer(
bootstrap_servers="localhost:9092",
value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)
order_id = 1000
while True:
order = {
"order_id": order_id,
"customer_id": random.randint(1, 5),
"total_amount": round(random.uniform(10, 500), 2),
}
producer.send("orders", key=str(order["customer_id"]).encode(), value=order)
print("sent", order)
order_id += 1
time.sleep(1)# the missing piece: writing each micro-batch to Postgres with an idempotent upsert
def write_batch_to_postgres(batch_df, batch_id):
(
batch_df.write
.format("jdbc")
.option("url", "jdbc:postgresql://localhost:5432/ecommerce")
.option("dbtable", "orders_stream")
.option("user", "postgres").option("password", "postgres")
.mode("append") # combine with an ON CONFLICT upsert at the DB level for true idempotency
.save()
)
query = (
orders.writeStream
.foreachBatch(write_batch_to_postgres) # gives you a plain DataFrame per micro-batch,
.outputMode("append") # so any batch-style write logic (Week 3) applies directly
.start()
)
query.awaitTermination()foreachBatch is the bridge between streaming and everything else on this site. Spark hands you a plain, regular DataFrame for each micro-batch, so the exact same SQLAlchemy/psycopg2 upsert patterns from Week 3 and Week 4's idempotency section work unchanged. Streaming doesn't need a whole new toolkit just for "write to a database" — only a different trigger for when that step runs.