Spark Streaming Kafka Offsets: Manual and Auto Management (2026)

Spark Streaming’s Kafka integration is a powerful tool for real-time data processing, but managing Kafka offsets can be a tricky business.

Let’s peek under the hood of a Spark Streaming job that’s pulling data from Kafka. Imagine this: your Spark job is merrily consuming messages from a Kafka topic, processing them, and writing the results somewhere else. Now, what happens when your Spark job restarts? How does it know where to pick up from in Kafka? This is where offset management comes in.

Here’s a simplified view of a Spark Streaming job reading from Kafka:

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.{OutputMode, Trigger}
import org.apache.spark.sql.types._

val spark = SparkSession.builder()
  .appName("KafkaOffsetExample")
  .master("local[*]") // For local testing
  .getOrCreate()

// Define the schema for your Kafka messages (adjust as needed)
val schema = StructType(Seq(
  StructField("message", StringType, true)
))

val kafkaStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092") // Your Kafka broker(s)
  .option("subscribe", "my_topic") // The Kafka topic to subscribe to
  .option("startingOffsets", "earliest") // Or "latest", or a specific offset JSON
  .load()

// Select the value from Kafka, cast it to string, and parse it if it's JSON
val processedStream = kafkaStream
  .selectExpr("CAST(value AS STRING)")
  // .select(from_json($"value", schema).as("data")) // Uncomment if your value is JSON
  // .select("data.message") // Uncomment if your value is JSON

// For demonstration, let's just print the messages to the console
val query = processedStream.writeStream
  .outputMode(OutputMode.Append)
  .format("console")
  .trigger(Trigger.ProcessingTime("5 seconds")) // Process data every 5 seconds
  .start()

query.awaitTermination()

This code sets up a Spark Streaming job to read from a Kafka topic named my_topic running on localhost:9092. The startingOffsets option dictates where Spark begins reading.

Auto Management: The "Set it and Forget It" Illusion

Spark Streaming, when configured correctly, can automatically manage Kafka offsets. This is usually done by leveraging Spark’s checkpointing mechanism. When Spark processes a batch of data from Kafka, it records the latest processed offset for each Kafka partition in its own checkpoint directory. Upon restart, Spark reads these saved offsets from the checkpoint and resumes consumption from the next message in each partition.

The key options for auto-management are:

• checkpointLocation: This is crucial. Spark writes its state, including Kafka offsets, to this directory. If it’s not set, Spark won’t remember where it left off.

  val query = processedStream.writeStream
    .outputMode(OutputMode.Append)
    .format("console")
    .option("checkpointLocation", "/path/to/your/checkpoint/dir") // REQUIRED for auto-management
    .trigger(Trigger.ProcessingTime("5 seconds"))
    .start()
  

  Why it works: Spark uses this distributed filesystem (like HDFS, S3, or even a local directory for testing) to store metadata about the streaming query, including the committed Kafka offsets. On restart, it reads this metadata to resume.

• startingOffsets: While checkpointLocation handles resuming, startingOffsets dictates where Spark begins if there’s no existing checkpoint or if you explicitly want to override it. Common values are "earliest" (start from the beginning of the topic) or "latest" (start from messages produced after Spark starts). You can also provide a JSON string for specific partition offsets.

  .option("startingOffsets", "earliest") // Or "latest", or a JSON like '{"partition":0,"offset":100}'
  

  Why it works: This option provides the initial instruction to Spark about where to begin reading before any checkpointed offsets are considered.

Manual Management: The "I Want Control" Approach

Sometimes, you need more granular control. This often involves disabling Spark’s automatic offset management and taking matters into your own hands, typically by using Kafka’s own offset management features or by manually committing offsets after your processing logic. This is less common with Spark Structured Streaming but was more prevalent in the older Spark Streaming (DStreams) API.

In Spark Structured Streaming, manual offset management is less about disabling Spark’s core mechanism and more about how you read and write. The most common scenario where you might feel you’re "manually" managing offsets is when you want to:

1. Re-process specific offsets: You might have a bug in your processing logic and need to re-run data from a particular point.
2. Consume from a specific point on startup: For debugging or recovery, you might want to force Spark to start from a known good offset.

The startingOffsets option is your primary tool here.

• Specific Offset JSON: You can provide a JSON string to startingOffsets to precisely control where Spark begins.

  // Example: Start from offset 150 in partition 0 and offset 200 in partition 1
  val specificOffsetsJson = """{"my_topic":{"0":150,"1":200}}"""
  val kafkaStream = spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "my_topic")
    .option("startingOffsets", specificOffsetsJson) // Manual starting point
    .load()
  

  Why it works: Spark parses this JSON and instructs the Kafka consumer to seek to these exact offsets before it starts reading any data.

• Using reset-offsets tool (for older Spark Streaming/DStreams or specific recovery scenarios): While not directly part of Structured Streaming’s readStream, for more complex recovery scenarios or if you’re dealing with older DStreams, you might interact with Kafka’s kafka-consumer-groups.sh tool. You can reset consumer group offsets to a specific point. If your Spark job is configured to use a specific Kafka consumer group ID, resetting the group offset in Kafka before starting Spark can influence where Spark begins, but this bypasses Spark’s internal checkpointing for the initial read.

  # Example to reset offsets for a consumer group
  kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my_spark_consumer_group --topic my_topic --reset-offsets --to-offset 100 --execute
  

  Why it works: This directly manipulates the offset committed by the consumer group in Kafka’s coordination topics. If Spark’s consumer group ID matches, Kafka will tell Spark that the last committed offset is the one you set. However, it’s generally recommended to rely on Spark’s checkpointLocation and startingOffsets for Structured Streaming to maintain a consistent state.

The Most Surprising Truth: Offsets Are Not Transactions

The most counterintuitive aspect of Kafka offset management, especially when integrating with systems like Spark, is that offsets are not transactional guarantees. Kafka guarantees that a message will be delivered at least once. Spark’s offset management, by default, aims for exactly-once processing semantics. It achieves this by writing the processed output and committing the Kafka offsets together atomically within Spark’s checkpointing mechanism. If Spark crashes after writing output but before committing offsets, upon restart, it will re-process the same messages. If it crashes before writing output but after committing offsets, you’ll lose those messages. The checkpointLocation is the linchpin for ensuring that Spark’s internal state (including committed offsets) is durable and consistent.

When you restart a Spark Streaming job, it first looks for a checkpointLocation. If found, it uses the committed offsets stored there to resume. If not found, or if startingOffsets is specified, it uses startingOffsets. The startingOffsets value is only consulted when there’s no existing checkpoint or when you explicitly want to override the resume point.

The next challenge you’ll likely encounter is handling data corruption or schema evolution in your Kafka messages.