Introduction

When Spark Streaming reads from Kafka with DStreams, two concerns usually matter together: how you filter records and when you consider offsets safe to commit. Filtering is just a transformation, but offset management is a correctness decision because it determines whether messages may be reprocessed or lost after failure.

Start with a Direct Stream

With Kafka-backed DStreams, the usual pattern is a direct stream. Filtering can then be expressed as a normal DStream transformation.

1import org.apache.kafka.common.serialization.StringDeserializer
2import org.apache.spark.streaming.kafka010._
3import org.apache.spark.streaming.{Seconds, StreamingContext}
4import org.apache.spark.{SparkConf, SparkContext}
5
6val conf = new SparkConf().setAppName("KafkaFilter")
7val sc = new SparkContext(conf)
8val ssc = new StreamingContext(sc, Seconds(5))
9
10val kafkaParams = Map[String, Object](
11  "bootstrap.servers" -> "localhost:9092",
12  "key.deserializer" -> classOf[StringDeserializer],
13  "value.deserializer" -> classOf[StringDeserializer],
14  "group.id" -> "stream-group",
15  "auto.offset.reset" -> "latest",
16  "enable.auto.commit" -> java.lang.Boolean.FALSE
17)
18
19val topics = Array("logs")
20val stream = KafkaUtils.createDirectStream[String, String](
21  ssc,
22  LocationStrategies.PreferConsistent,
23  ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
24)
25
26val errors = stream.filter(record => record.value.contains("ERROR"))
27errors.foreachRDD(rdd => println(s"error count = ${rdd.count()}"))

The filtering part is easy. The harder part is deciding when offsets should advance.

Commit Offsets Only After Successful Processing

A safe rule is: commit offsets only after the batch has been processed successfully. That way, if the job crashes mid-batch, Kafka records can be replayed.

1stream.foreachRDD { rdd =>
2  if (!rdd.isEmpty()) {
3    val offsets = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
4
5    val filtered = rdd.filter(record => record.value.contains("ERROR"))
6    filtered.foreachPartition { partition =>
7      partition.foreach(record => println(record.value))
8    }
9
10    stream.asInstanceOf[CanCommitOffsets].commitAsync(offsets)
11  }
12}

The important sequencing is:

1. get the offset ranges for the batch
2. process the data
3. commit only after processing succeeds

That ordering is what gives you at-least-once behavior.

Filtering Does Not Change the Commit Principle

A common mistake is thinking that only filtered records matter for offsets. Kafka offsets belong to the consumed batch, not just to the records you kept after filtering.

So even if the filter drops ninety percent of the messages, the commit decision is still based on whether the whole consumed batch was handled correctly according to your logic.

Think About Your Processing Guarantees

With manual commits after successful processing, the typical result is at-least-once processing. That means duplicates are possible after failure and replay.

If duplicates matter, make the downstream effect idempotent. For example:

• write with upsert semantics
• deduplicate by message key
• make sinks replay-safe

Offset management alone does not create exactly-once semantics for arbitrary side effects.

DStreams Are a Legacy API

Spark's DStream API still appears in existing systems, but newer Spark development often prefers Structured Streaming. If you are maintaining an older DStream application, understand the offset and failure model clearly rather than copying modern Structured Streaming assumptions into it.

If your sink writes to an external database, treat offset commit as the final acknowledgement step after the write succeeds. That sequencing is the difference between tolerable replay and silent data loss, and it is the heart of correct offset management in older DStream-based systems.

Common Pitfalls

• Committing offsets before the batch is fully processed.
• Assuming filtered-out records do not affect offset progression.
• Enabling Kafka auto-commit while also trying to manage offsets manually.
• Ignoring duplicate processing risk after a crash and replay.
• Mixing DStream assumptions with Structured Streaming semantics.

Summary

• Filtering DStreams is straightforward, but offset management determines correctness.
• Use a direct stream and disable Kafka auto-commit when managing offsets yourself.
• Commit offsets only after successful batch processing.
• Filtering does not change the fact that offsets belong to the consumed batch.
• DStreams are older Spark APIs, so be explicit about the delivery guarantees you want.