在Spark Streaming中，如何检测空批处理？

Question

在Spark Streaming中，如何检测空批处理？

让我们看看有状态的流式字数统计例子：https://github.com/apache/spark/blob/master/examples/src/main/java/org/apache/spark/examples/streaming/JavaStatefulNetworkWordCount.java。只有在将新单词添加到流中时，是否可以打印单词计数RDD？

Answer 1

以下是我如何做到的。创建一个空的RDD，它是您的previousWindow。然后在forEachRDD中，计算最后一个窗口和当前窗口之间的差异。如果当前窗口包含的记录不在以前的窗口中，则该批次中有新内容。最后，将上一个窗口设置为当前窗口中的内容。

... 

    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 
}