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

6
votes

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?

2
apache-spark

2 Answers

2
votes

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
}
0
votes

This is how i avoid empty batches and overwrite in same directory.

import java.time.format.DateTimeFormatter
import java.time.LocalDateTime

   messageRecBased.foreachRDD{ rdd =>
        rdd.repartition(1)
        val eachRdd = rdd.map(record => record.value)
        if(!eachRdd.isEmpty)
          eachRdd.saveAsTextFile("hdfs/location/"+DateTimeFormatter.ofPattern("yyyyMMddHHmmss").format(LocalDateTime.now)+"/")
      }