How do I read from Kafka in Structured Streaming?

Summary

• Read Kafka with spark.readStream.format("kafka"), set kafka.bootstrap.servers, and choose exactly one of subscribe, subscribePattern or assign.
• Each row has key, value, topic, partition, offset, timestamp and timestampType columns, plus headers when includeHeaders is true. Keys and values are always binary, so cast or parse them yourself.
• startingOffsets defaults to latest for streaming queries and applies only when a new query starts. A restarted query picks up where it left off.
• maxOffsetsPerTrigger caps how many offsets each micro-batch reads, and you can add or change it between restarts.
• failOnDataLoss defaults to true. Spark does not commit offsets back to Kafka, and it does not reset them when Kafka removes data.

How do I read from Kafka in Structured Streaming?

How do I read from Kafka in Structured Streaming?

Use spark.readStream with the kafka format, pass the broker list in kafka.bootstrap.servers, and select the topics to read. The Spark Kafka source works with Kafka brokers version 0.10.0 or higher. Python applications must add the Kafka connector package when they launch (use the version that matches your Spark release):

./bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0 my_stream.py

Which option selects the topics?

Specify exactly one of the three options below.

Option	Value	Reads
subscribe	Comma-separated topic list	The named topics
subscribePattern	Java regex string	Every topic matching the pattern
assign	JSON such as {"topicA":[0,1],"topicB":[2,4]}	Only the listed partitions

What does a basic query look like?

df = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
    .option("subscribe", "topic1")
    .load())

df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

Each row has these columns: key (binary), value (binary), topic (string), partition (int), offset (long), timestamp (timestamp), timestampType (int), and an optional headers (array) when includeHeaders is true. The source always uses ByteArrayDeserializer, so deserialize with DataFrame operations. To parse JSON, cast to string first:

from pyspark.sql.functions import from_json, col

parsed = (df.select(from_json(col("value").cast("string"),
                              "event_type STRING, event_ts TIMESTAMP").alias("value"))
            .select("value.*"))

Where does the query start, and how does it resume?

startingOffsets accepts earliest, latest, or a JSON string with an offset per partition, such as {"topic1":{"0":23,"1":-2}}, where -2 means earliest and -1 means latest. The default is latest for streaming queries and earliest for batch queries. For a streaming query it "only applies when a new query is started, and that resuming will always pick up from where the query left off. Newly discovered partitions during a query will start at earliest."
Spark tracks progress in the query checkpoint, not in Kafka. The docs say the Kafka source "doesn't commit any offset," and each query generates its own consumer group id by default. Because Spark manages offsets itself, you cannot set the Kafka auto.offset.reset parameter. Use startingOffsets instead.

How do I limit how much each micro-batch reads?

• maxOffsetsPerTrigger is a "rate limit on maximum number of offsets processed per trigger interval." Spark splits the total proportionally across topic partitions of different volume. There is no limit by default.
• minPartitions raises the number of Spark partitions. By default each Kafka topic partition maps to one Spark partition, and a larger value splits big Kafka partitions into smaller pieces.
• minOffsetsPerTrigger waits for a minimum number of offsets before a trigger fires, and maxTriggerDelay (default 15m) bounds that wait.

Adding, deleting or modifying rate limits is allowed when you restart a query from the same checkpoint.

What happens when Kafka deletes data I have not read yet?

failOnDataLoss (default true) controls "whether to fail the query when it's possible that data is lost (e.g., topics are deleted, or offsets are out of range)." The docs add that this "may be a false alarm" and that you can disable it. Spark does not reset offsets to recover. When the offsets a query consumed no longer exist in Kafka, the application "will see data loss," and in extreme cases the input rows of a batch can shrink to zero. Enabling failOnDataLoss makes Structured Streaming fail the query for such cases.

How do I handle late or out-of-order messages?

Event time is the time embedded in the data itself, so you choose the timestamp column: the Kafka timestamp column or a field parsed from the payload. Call withWatermark(eventTime, delayThreshold) before the aggregation. Data that arrives within the watermark delay is guaranteed to be aggregated. Data delayed by more than the threshold is not guaranteed to be dropped: it "may or may not get aggregated."

Can I change the subscription of a running query?

Not safely. The Spark docs say changes to subscribed topics "are generally not allowed as the results are unpredictable," and the Databricks docs list subscribed Kafka topics among the changes that generally require a new checkpoint. A new checkpoint location starts the query fresh, so startingOffsets applies again.

How does Kafka reading work on Databricks?

Use the same readStream code. In Lakeflow Pipelines or streaming tables in Databricks SQL, the read_kafka function (Databricks Runtime 13.3 LTS and above) reads a topic:

CREATE OR REFRESH STREAMING TABLE catalog.schema.raw_events AS
SELECT * FROM STREAM read_kafka(
  bootstrapServers => 'kafka_server:9092',
  subscribe => 'events'
);

For incremental batch loading, Databricks recommends Kafka with Trigger.AvailableNow. Authentication options include Unity Catalog service credentials and SASL/SSL. To watch backlog, read avgOffsetsBehindLatest, maxOffsetsBehindLatest, minOffsetsBehindLatest and estimatedTotalBytesBehindLatest in the query progress.

FAQs

Why do my key and value columns look like bytes?

They are always deserialized as byte arrays. Cast them to string, or use a function such as from_avro, to read them.

Does Spark use Kafka consumer groups to remember its place?

No. Offsets live in the checkpoint, and the source generates a unique group id per query so each source reads all partitions of its subscribed topics without interference from other consumers.