Introduction

When you create a Kafka InputDStream with KafkaUtils.createDirectStream, offsets determine where consumption begins for each topic partition. The direct stream approach is useful because Spark reads directly from Kafka partitions instead of relying on receivers, which makes offset handling more explicit and reliable.

Build the Direct Stream with Explicit Starting Offsets

In the older Spark Streaming Kafka integration, you can pass a fromOffsets mapping to start from chosen offsets per topic partition.

1from pyspark import SparkContext
2from pyspark.streaming import StreamingContext
3from pyspark.streaming.kafka import KafkaUtils, TopicAndPartition
4
5sc = SparkContext(appName="KafkaDirectStreamWithOffsets")
6ssc = StreamingContext(sc, 10)
7
8kafka_params = {
9    "metadata.broker.list": "localhost:9092",
10    "auto.offset.reset": "smallest",
11}
12
13from_offsets = {
14    TopicAndPartition("events", 0): 50,
15    TopicAndPartition("events", 1): 120,
16}
17
18stream = KafkaUtils.createDirectStream(
19    ssc,
20    topics=["events"],
21    kafkaParams=kafka_params,
22    fromOffsets=from_offsets,
23)

This tells Spark exactly where to start for each partition instead of relying entirely on the default offset reset policy.

Process the Stream and Inspect Offset Ranges

A direct stream lets you inspect offset ranges per RDD batch, which is useful for logging and manual offset management patterns.

1def process_rdd(time, rdd):
2    if rdd.isEmpty():
3        return
4
5    offset_ranges = rdd.offsetRanges()
6    for offset in offset_ranges:
7        print(
8            offset.topic,
9            offset.partition,
10            offset.fromOffset,
11            offset.untilOffset,
12        )
13
14    records = rdd.map(lambda msg: msg[1]).collect()
15    print(records)
16
17
18stream.foreachRDD(process_rdd)
19ssc.start()
20ssc.awaitTermination()

The exact batch contents and offset ranges now move together, which makes replay and monitoring easier.

Recover from Stored Offsets Carefully

In many legacy DStream applications, offsets are read from an external store such as a database or coordination table during startup, then written back only after processing completes successfully. That pattern can work, but only if the offset write is treated as part of successful batch completion.

If the application crashes after storing the new offsets but before finishing the downstream write, the next restart can skip data. If it crashes before storing the offsets, the next restart may replay the batch. That is why offset ownership has to be designed alongside output semantics, not as an isolated startup parameter.

Think Carefully About Offset Ownership

Choosing the starting offsets is only one part of the problem. You also need a strategy for where offsets come from and when they should advance.

Common patterns include:

• starting from Kafka defaults for new jobs
• reading offsets from an external store after successful processing
• replaying from a known offset for recovery or backfill

If your application manages offsets outside Kafka consumer groups, make sure offsets are updated only after the batch has been processed successfully. Otherwise the job may skip data after failures.

This matters especially with older DStream-based code, where offset management was often designed manually around batch completion.

Common Pitfalls

The biggest mistake is assuming auto.offset.reset controls everything even when you are supplying explicit fromOffsets. Once you pass explicit offsets, those values are the real starting point.

Another common issue is forgetting that offsets are tracked per topic partition, not just per topic. If one partition is missing from your mapping, startup behavior can become inconsistent.

People also move offsets forward before the output is safely committed. That creates data loss during failure recovery because the stream appears to have consumed records that were never fully processed.

Finally, remember that this API belongs to the older Spark Streaming DStream model. If you are starting a new project, Structured Streaming is usually the more modern choice.

Summary

• Use fromOffsets with TopicAndPartition keys to start a direct Kafka stream from explicit offsets.
• Inspect offsetRanges() inside foreachRDD to see what each batch consumed.
• Manage offsets per partition, not just per topic.
• Advance stored offsets only after successful processing.
• For new systems, prefer Structured Streaming unless you are maintaining an existing DStream pipeline.