Spark Streaming：如何通过 StreamingListener 获取处理时间和调度延迟？

Question

官方文档说我们可以通过 StreamingListener 获取处理时间的值：Spark Streaming 程序的进度也可以通过 StreamingListener 接口进行监控，可以获取接收器状态和处理时间。 http://spark.apache.org/docs/latest/streaming-programming-guide.html#monitoring-applications

我知道有一些关于 Spark 的指标的信息，但它不包含处理时间和调度延迟。 http://spark.apache.org/docs/latest/monitoring.html#rest-api

我阅读了StreamingListener的源代码。它包含一个这样的方法：

def printStats() {
    showMillisDistribution("Total delay: ", _.totalDelay)
    showMillisDistribution("Processing time: ", _.processingDelay)
}

我认为可以获得这些指标是可能的，但我没有意识到这一点。 我需要这些指标来进行研究。我怎样才能得到它们？非常感谢。

Answer 1

我找到了解决办法。

class MyListener() extends StreamingListener {
    override def onBatchCompleted(batchStarted: StreamingListenerBatchCompleted) {
    println("Total delay: " + batchStarted.batchInfo.totalDelay)
    println("Processing time: " + batchStarted.batchInfo.processingDelay)
  }
}