Kafka with PySpark Structured Streaming: read a topic into a Delta table

A stream is data that never ends: orders placed, payments taken, sensors read, each an event as it happens. Module 6 streamed files into a table with a checkpoint and availableNow. This module streams events from Kafka, the most common place they wait, then counts them by when they happened, makes sure each is written once, and asks when a stream is worth it at all. The streaming conceptStreamingGetting a model's answer in small pieces as it is written, instead of all at once at the end, so people see text start sooner and long answers do not hit a timeout. covers the ideas; this part gets events from Kafka into a Delta table.

A topic, its partitions and offsets

Kafka keeps events in topics, such as orders. A topic is split into partitions, so several machines can write and read it at once. An event goes to 1 partition and gets the next offset there, its position counting from 0. Events with the same key go to the same partition, and each partition keeps its events in the order they were written. Kafka does not delete an event when it is read; a topic keeps events for as long as its retention setting says, so several readers can each read the same events at their own pace.

So a reader is a set of positions, 1 per partition: the next offset it will read. Spark keeps those positions in the stream's checkpoint.

Kafka on your computer

Start a single Kafka node in Docker, which is both the broker that stores events and the controller that manages it, and create a topic with 3 partitions:

docker run -d --name kafka -p 127.0.0.1:9092:9092 \
  -e KAFKA_NODE_ID=1 -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  apache/kafka:4.3.1
docker exec --workdir /opt/kafka/bin kafka ./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3 --replication-factor 1

The second command prints Created topic orders. Spark reads Kafka through a connector that is not in PySpark itself. Save this as stream_session.py; every script in the module starts its session with it, and it downloads the connector the first time:

"""A local Spark session with Delta Lake and Spark's Kafka connector. On Databricks, use the spark it gives you."""
from delta import configure_spark_with_delta_pip
from pyspark.sql import SparkSession

KAFKA_CONNECTOR = "org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0"


def stream_spark(app: str = "stream") -> SparkSession:
    builder = (
        SparkSession.builder.master("local[4]").appName(app)
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
        # A few partitions for the state a stream keeps: the default of 200 is for a cluster.
        .config("spark.sql.shuffle.partitions", "4")
        .config("spark.log.level", "ERROR")
        .enableHiveSupport()
    )
    return configure_spark_with_delta_pip(builder, extra_packages=[KAFKA_CONNECTOR]).getOrCreate()

The connector's version must match your PySpark, here 4.2.0; pip install pyspark==4.2.0 delta-spark as in module 4.

Send events

The shop's system would send its own events. To stand in for it, Spark can write to Kafka too: a DataFrame with a key and a value column, written in the kafka format. Save this as make_orders.py:

"""Send 1 batch of the shop's order events to the Kafka topic orders, as the shop's system would."""
import sys
from pyspark.sql import functions as F
from stream_session import stream_spark

BATCHES = {
    "1": [
        (1001, "C01", "placed", 25.50, "2026-10-01T09:15:00Z"),
        (1002, "C02", "placed", 12.00, "2026-10-01T09:40:00Z"),
        (1003, "C03", "placed", 31.00, "2026-10-01T09:55:00Z"),
    ],
    "2": [
        (1001, "C01", "paid", 25.50, "2026-10-01T10:05:00Z"),
        (1004, "C01", "placed", 40.00, "2026-10-01T10:20:00Z"),
    ],
}

spark = stream_spark("make_orders")
events = spark.createDataFrame(
    BATCHES[sys.argv[1]], "order_id INT, customer_id STRING, status STRING, amount DOUBLE, event_time STRING"
)
# Kafka stores a key and a value as bytes. The key is the order, so all its events go to 1 partition, in order.
(
    events.select(F.col("order_id").cast("string").alias("key"), F.to_json(F.struct("*")).alias("value"))
    # 1 task sends the batch, in the order listed; several would send at once, in any order.
    .coalesce(1)
    .write.format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("topic", "orders")
    .save()
)
print("sent", events.count(), "events")

python make_orders.py 1 prints sent 3 events. Each value is a line of JSON, such as {"order_id":1001,"customer_id":"C01","status":"placed","amount":25.5,"event_time":"2026-10-01T09:15:00Z"}, and event_time is when the order was placed, which part 2 counts by.

Read the topic into a table

Save this as read_orders.py:

"""Load the new events on the topic orders into the Delta table bronze_orders, then stop."""
from pyspark.sql import functions as F
from stream_session import stream_spark

SCHEMA = "order_id INT, customer_id STRING, status STRING, amount DECIMAL(10,2), event_time TIMESTAMP"

spark = stream_spark("read_orders")
raw = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "orders")
    .option("startingOffsets", "earliest")
    .load()
)
events = raw.select(
    F.from_json(F.col("value").cast("string"), SCHEMA).alias("event"),
    "partition",
    "offset",
).select("event.*", "partition", "offset")

query = (
    events.writeStream.format("delta")
    .option("checkpointLocation", "checkpoints/bronze_orders")
    .trigger(availableNow=True)
    .toTable("bronze_orders")
)
query.awaitTermination()
print("loaded", sum(progress.numInputRows for progress in query.recentProgress), "events")

The Kafka source gives every event as a row with key and value as bytes, plus topic, partition, offset and Kafka's own timestamp. The value is cast to a string and parsed with a schema, and partition and offset are kept, so every row in the table says where in Kafka it came from. availableNow reads what is on the topic, writes it, and stops, as module 6's file stream did. Save a small show.py too:

"""Show a table, sorted by all its columns, so it prints the same every time."""
import sys
from stream_session import stream_spark

spark = stream_spark("show")
table = spark.table(sys.argv[1])
table.orderBy(*table.columns).show(truncate=False)

python read_orders.py prints loaded 3 events. Send the second batch and read again, twice:

python make_orders.py 2
python read_orders.py
python read_orders.py

The first read prints loaded 2 events, the second loaded 0 events. python show.py bronze_orders:

+--------+-----------+------+------+-------------------+---------+------+
|order_id|customer_id|status|amount|event_time         |partition|offset|
+--------+-----------+------+------+-------------------+---------+------+
|1001    |C01        |paid  |25.50 |2026-10-01 10:05:00|0        |2     |
|1001    |C01        |placed|25.50 |2026-10-01 09:15:00|0        |0     |
|1002    |C02        |placed|12.00 |2026-10-01 09:40:00|0        |1     |
|1003    |C03        |placed|31.00 |2026-10-01 09:55:00|1        |0     |
|1004    |C01        |placed|40.00 |2026-10-01 10:20:00|1        |1     |
+--------+-----------+------+------+-------------------+---------+------+

Both events of order 1001 are in partition 0, placed at offset 0 and paid at offset 2, because they share a key. Partition 2 got nothing: which partition a key lands in depends on a hash of the key, not on an even share-out.

What the checkpoint remembers

checkpoints/bronze_orders/offsets/ has a file per batch. The last line of each is where that batch read to, per partition. After the first batch, offsets/0 ends with:

{"orders":{"0":2,"1":1,"2":0}}

and after the second, offsets/1 ends with {"orders":{"0":3,"1":2,"2":0}}. These are the next offsets to read: partition 0 has offsets 0 to 2, so the next is 3. Spark writes this before it runs a batch and marks the batch done in commits/ after, which is what part 3 relies on.

startingOffsets only matters the first time. Spark's docs say it applies when a new query starts, and a query that resumes always continues from its checkpoint, so changing it later does nothing. 2 other options matter in production: maxOffsetsPerTrigger caps how many events a batch reads, so a stream that falls behind catches up in steps, and failOnDataLoss, on by default, stops the stream when the offsets it expects are gone, for example because the topic's retention deleted them before the stream read them.

The number of partitions also decides how much of the work runs in parallel: Spark reads each Kafka partition with 1 task, unless minPartitions asks it to split them. How many a topic needs is a sizing question:

ToolKafka sizing calculatorMessages, their size, retention and readers in; the partitions, disk, network and brokers out.

Part 2 counts these events by when they happened, including the ones that arrive late.