In Spark Streaming, how to detect for an empty batch?

Question

In Spark Streaming, how to detect for an empty batch?

Let's take the stateful streaming word count example: https://github.com/apache/spark/blob/master/examples/src/main/java/org/apache/spark/examples/streaming/JavaStatefulNetworkWordCount.java. Is it possible to print the word count RDD only when new words are added to the stream?

Answer 1

Here's how I did it. Create an empty RDD that is your previousWindow. Then in the forEachRDD, compute the difference between the last window and the current window. if the current window contains records not in the previous window, there is something new in the batch. Finally, set the previous window to what is in the current window.

  ...

  var previousWindowRdd = sc.emptyRDD[String]

  dStream.foreachRDD {
    windowRdd => {
      if (!windowRdd.isEmpty) processWindow(windowRdd.cache())
    }
  }

  ...

def processWindow(windowRdd: RDD[String]) = {
  val newInBatch = windowRdd.subtract(previousWindowRdd)

  if (!newInBatch.isEmpty())
    processNewBatch(windowRdd)

  previousWindowRdd = windowRdd
}