Pause Spark Streaming Job

Question

I have simple spark streaming application which reads data from Kafka and then send this data after transformation on a http end point (or another kafka - for this question let's consider http). I am submitting jobs using job-server.

I am currently starting the consumption from source kafka with "auto.offset.reset"="smallest" and interval=3s. In happy case everything looks good. Here's an excerpt:

kafkaInputDStream.foreachRDD(rdd => {
  rdd.foreach(item => {
  //This will throw exception if http endpoint isn't reachable
      httpProcessor.process(item._1, item._2)
  })
})

Since "auto.offset.reset"="smallest", this processes about 200K messages in one job. If I stop http server mid-job (simulating some issue in POSTing) and httpProcessor.process throws exception, that Job fails and whatever is unprocessed is lost. I see it keeps on polling every 3 seconds after that.

So my question is:

1. Is my assumption right that if in next 3 second job if it got X messages and only Y could be processed before hitting an error, rest X-Y will not be processed?
2. Is there a way to pause the stream/consumption from Kafka? For instance in case there's a intermittent network issue and most likely all messages consumed will be lost in that time. Something which keeps on retrying (maybe exponential backoff) and whenever http end point is UP, start consuming again.

Thanks

Answer 1

Yes, your assumption is correct that if your partition fails, the remaining events are not processed for the moment.

However, there are quite a few parameter for you to tune to get your desired behavior (if you use DirectKafkaInputDStream).

Lets start with "auto.offset.reset"="smallest": This parameter tells Kafka to begin from the beginning, when there is no stored commit for the current group. As you mentioned that your RDD contains a lot of messages after starting, I assume that you do not commit your messages properly. If you expect exactly-once semantics, you definitely should consider keep track of your offsets as the DirectKafkaStreamInput explicitly does not keep track of that.

Starting offsets are specified in advance, and this DStream is not responsible for committing offsets, so that you can control exactly-once

Comment in the DirectKafkaInputSream Branch 1.6

That said currently your message are reprocessed, every time you restart your streaming job.

If you commit your processed offsets and pass it into the InputDStream on startup, the listener will continue from the last committed offset.

Regarding backpressure, the DirectKafkaInputDStream already uses a RateController which estimates how much events should be processed in one batch.

To use it, you have to enable backpressure:

"spark.streaming.backpressure.enabled": true

You can also limit the "spark.streaming.kafka.maxRatePerPartition" to add a upper bound for the batch size.

If you want to control backpressure on your own (and perhaps stop the consumer completely for a while), you may want to implement some methods of StreamingListener and use it in your job. You can e.g. decide after each completed batch to stop your streaming job or not with the StreamingListener.

Answer 2

I think Spring Cloud Stream could solve your problem. Kafka is source. Spark Streaming is processor. Http is sink. Only if there is input from Kafka, Spark Streaming will process. You needn't stop or resume input from Kafka. Hope it helps.