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

import org.apache.spark.streaming.scheduler.{StreamingListener, StreamingListenerBatchCompleted}

class CustomStreamingListener extends StreamingListener {
  override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {
    val batchInfo = batchCompleted.batchInfo
    val processingTime = batchInfo.processingDelay
    val schedulingDelay = batchInfo.schedulingDelay

    // 处理时间和调度延迟的处理逻辑
    // ...

    println(s"Processing time: $processingTime ms")
    println(s"Scheduling delay: $schedulingDelay ms")
  }
}

// 创建StreamingContext和DStream等代码省略

// 创建自定义StreamingListener对象
val customListener = new CustomStreamingListener

// 将自定义StreamingListener对象注册到StreamingContext中
streamingContext.addStreamingListener(customListener)

// 启动StreamingContext
streamingContext.start()
streamingContext.awaitTermination()