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

Spark reads Kafka with the kafka source: spark.readStream.format("kafka") with the brokers' addresses and the topic. Each row has the event's key and value as bytes, its topic, partition and offset, so the value is cast to a string and parsed, for example with from_json. The stream writes to a Delta table with a checkpoint, which records the next offset of each partition after every batch, so a rerun reads only events that arrived since. trigger(availableNow=True) reads what is there and stops.

- Page: https://tecyfy.com/data-engineering/course/streaming
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 9: Streaming, Kafka into a Delta table
- Parts of this module: Kafka into a Delta table (https://tecyfy.com/data-engineering/course/streaming); Event time, windows and watermarks (https://tecyfy.com/data-engineering/course/streaming/event-time); Exactly-once and foreachBatch (https://tecyfy.com/data-engineering/course/streaming/exactly-once); Streaming on Databricks (https://tecyfy.com/data-engineering/course/streaming/on-databricks); Build it: orders from Kafka to an hourly table (https://tecyfy.com/data-engineering/course/streaming/build-it)

At the end of this part you can:

- Run Kafka on your computer, create a topic and send JSON events to it with a key
- Read a topic with Structured Streaming, parse its values and write them to a Delta table with a checkpoint
- Read what a checkpoint records for Kafka, and say which option decides where a stream starts and when it no longer does

A stream is data that never ends: orders placed, payments taken, sensors read, each an event as it happens. [Module 6](/data-engineering/course/getting-data-in) 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 concept](https://tecyfy.com/concepts/streaming) 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:

```bash
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:

```python
"""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](/data-engineering/course/delta-lake-and-iceberg).

## 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`:

```python
"""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`:

```python
"""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:

```python
"""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:

```bash
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`:

```text
+--------+-----------+------+------+-------------------+---------+------+
|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:

```text
{"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:

- Tool: [Kafka sizing calculator](https://tecyfy.com/data-engineering/kafka-sizing). Messages, their size, retention and readers in; the partitions, disk, network and brokers out.

[Part 2](/data-engineering/course/streaming/event-time) counts these events by when they happened, including the ones that arrive late.

## Check yourself

1. You change startingOffsets from earliest to latest and run read_orders.py again with the same checkpoint. What does it read?

   A. Only events sent after this run starts
   B. The events after the offsets in the checkpoint, as before, because startingOffsets applies only to a new query
   C. Every event on the topic again
   D. Nothing, because the option changed and the query fails

2. The 2 events for order 1001 both landed in partition 0. Why can you rely on that?

   A. Spark sorts events by order_id before writing
   B. Partition 0 is the default for every event
   C. Kafka sends events with the same key to the same partition, and keeps each partition in the order the events were written
   D. Because they arrived in the same second

### Answers

1. B. Spark's docs say startingOffsets applies only when a query starts for the first time; a query that resumes always picks up where its checkpoint left off. To start from somewhere else, you start a new query with a new checkpoint, and decide what that means for the table it writes to.
2. C. make_orders.py uses the order id as the key, and Kafka writes events with the same key to the same partition and keeps each partition in order. So a reader sees placed before paid for each order. Across partitions there is no order, which is why events for different orders can interleave.

## Questions

### How do I read from Kafka with PySpark?

Add Spark's Kafka connector, org.apache.spark:spark-sql-kafka-0-10_2.13 for your Spark version, then use spark.readStream.format("kafka") with kafka.bootstrap.servers and subscribe. The key and value come as bytes, so cast the value to a string and parse it, with from_json for JSON. Write the result with a checkpointLocation so the stream remembers its offsets. On Databricks the connector is already there.

### What is an offset in Kafka?

An offset is an event's position in its partition, counting from 0. Each partition of a topic has its own offsets, and a reader keeps track of the next offset to read in each partition. Spark Structured Streaming keeps those offsets in its checkpoint, not in Kafka, and records the range it is about to read before each batch.

## Sources

- [Introduction](https://kafka.apache.org/intro)
- [Producer configs](https://kafka.apache.org/43/configuration/producer-configs/)
- [Quickstart](https://kafka.apache.org/quickstart)
- [apache/kafka](https://hub.docker.com/r/apache/kafka)
- [Structured Streaming + Kafka Integration Guide](https://spark.apache.org/docs/latest/streaming/structured-streaming-kafka-integration.html)
- [Structured Streaming Programming Guide, getting started](https://spark.apache.org/docs/latest/streaming/getting-started.html)
- [Connect to Apache Kafka](https://docs.databricks.com/aws/en/connect/streaming/kafka)
- [Configure Structured Streaming trigger intervals](https://docs.databricks.com/aws/en/structured-streaming/triggers)