Track of consumed messages in Spark structured streaming

Question

I want to set config where my application keep track of consumed messages from kafka. So that whenever it gets failed, it could pick from the last commit or consumed offset onwards.

readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load()
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "topic1")
  .trigger(Trigger.Continuous("1 second"))  // only change in query
  .start();

I've read online that checkpointlocation property can be set which can used by spark to keep track of offsets.

Want to know where I can set this property ? can I set in above code within option ? May I know how can I set it properly.

secondly, I'm not able to understand trigger(Trigger.Continuous("1 second")) property. Docs says continuous processing engine will record the progress of the query every second, what kind of progress it record while reading messages from kafka ?

user_12208910 · Answer

You can set the checkpoint location as an option in your writeStream:

[...]
.writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "topic1")
  .option("checkpointLocation", "/path/to/dir")
  .trigger(Trigger.Continuous("1 second"))
  .start();

Tracking progress when reading from Kafka means to keep track of the consumed offsets within a TopicPartition. Setting a checkpoint location will enable your application to store that information in the given path as an JSON object, e.g.

{
  "topic1":{
    "0":11, 
    "1":101
  }
}

which means that the application has already consumed offset 10 in partition 0 and offset 100 in partition 1 of the topic topic1. The checkpoints are written "ahead" (using write-ahead-logs), so the application will continue to read messages from Kafka where it left off before the intended or unintended (failure) restart.

The Trigger.Continuous is available since Spark version 2.3. and as of now marked as experimental. Compared to the micro-batch approach, it will fetch every single message in Kafka as soon as it arrives in the topic without trying to batch it with other messages. This can improve latency but will most likely reduce your overall throughput.

The argument (e.g. 1 seconds) determines the frequency of checkpointing.

When using this trigger mode it is important to have at least as many cores available as the topic has partitions. Otherwise, the application will not make any progress. You can read more about it here:

"For example, if you are reading from a Kafka topic that has 10 partitions, then the cluster must have at least 10 cores for the query to make progress."

Track of consumed messages in Spark structured streaming

1 Answers