made changes to capture job level jobCreateDelay metric

Author: agsachin

external/kafka/src/test/scala/org/apache/spark/streaming/kafka/DirectKafkaStreamSuite.scala

@@ -19,28 +19,26 @@ package org.apache.spark.streaming.kafka
 
 import java.io.File
 import java.util.Arrays
-import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.ConcurrentLinkedQueue
-
-import scala.collection.JavaConverters._
-import scala.concurrent.duration._
-import scala.language.postfixOps
+import java.util.concurrent.atomic.AtomicLong
 
 import kafka.common.TopicAndPartition
 import kafka.message.MessageAndMetadata
 import kafka.serializer.StringDecoder
-import org.scalatest.{BeforeAndAfter, BeforeAndAfterAll}
-import org.scalatest.concurrent.Eventually
-
-import org.apache.spark.{SparkConf, SparkContext, SparkFunSuite}
 import org.apache.spark.internal.Logging
 import org.apache.spark.rdd.RDD
-import org.apache.spark.streaming.{Milliseconds, StreamingContext, Time}
 import org.apache.spark.streaming.dstream.DStream
-import org.apache.spark.streaming.kafka.KafkaCluster.LeaderOffset
 import org.apache.spark.streaming.scheduler._
 import org.apache.spark.streaming.scheduler.rate.RateEstimator
+import org.apache.spark.streaming.{Seconds, Milliseconds, StreamingContext, Time}
 import org.apache.spark.util.Utils
+import org.apache.spark.{SparkConf, SparkContext, SparkFunSuite}
+import org.scalatest.concurrent.Eventually
+import org.scalatest.{BeforeAndAfter, BeforeAndAfterAll}
+
+import scala.collection.JavaConverters._
+import scala.concurrent.duration._
+import scala.language.postfixOps
 
 class DirectKafkaStreamSuite
   extends SparkFunSuite
@@ -479,6 +477,201 @@ class DirectKafkaStreamSuite
   }
 }
 
+object DirectKafkaWordCountLocal {
+
+  import org.apache.spark.streaming._
+
+  case class Tick(symbol: String, price: Int, ts: Long)
+
+  def main(args: Array[String]) {
+
+    // Create context with 2 second batch interval
+    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("DirectKafkaWordCount")
+    val ssc = new StreamingContext(sparkConf, Seconds(2))
+    ssc.checkpoint("/tmp/checkpoint")
+    val listener = new LatencyListener(ssc)
+    ssc.addStreamingListener(listener)
+    val lines = ssc.socketTextStream("localhost", 8888)
+
+    val words = lines.flatMap(_.split(" "))
+
+    val pairs = words.map(word => (word, 1))
+
+    val wordCounts = pairs.reduceByKeyAndWindow(_+_, _-_, Seconds(60),Seconds(10))
+
+
+    val wordCountNew = wordCounts.filter(_._1.startsWith("sac")).reduceByKeyAndWindow(_+_, _-_, Seconds(60),Seconds(10))
+    wordCountNew.print()
+
+    ssc.start()
+    ssc.awaitTermination()
+  }
+}
+object DirectKafkaWordCountKafka {
+
+  import org.apache.spark.streaming._
+
+  def main(args: Array[String]) {
+    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("DirectKafkaWordCount")
+
+    val ssc = new StreamingContext(sparkConf, Seconds(2))
+    //ssc.checkpoint(checkPointPath)
+
+    val listener = new LatencyListener(ssc)
+    ssc.addStreamingListener(listener)
+    val kafkaBrokers = "localhost"
+    val kafkaPort ="9092"
+    val topic="test"
+
+    val topicsSet = Set(topic)
+
+    val brokerListString = new StringBuilder();
+
+    brokerListString.append(kafkaBrokers).append(":").append(kafkaPort)
+
+
+    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokerListString.toString())
+    System.err.println(
+      "Trying to connect to Kafka at " + brokerListString.toString())
+    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
+      ssc, kafkaParams, topicsSet)
+    ssc.checkpoint("/tmp/checkPoint/")
+    // Create context with 2 second batch interval
+
+    //val lines = ssc.socketTextStream("localhost", 9998)
+
+    val words = messages.map(x => x._2).flatMap(_.split(" "))
+
+    val pairs = words.map(word => (word, 1))
+    pairs.print()
+
+    val wordCounts = pairs.reduceByKeyAndWindow(_+_, _-_, Seconds(60),Seconds(10))
+
+    wordCounts.print()
+
+    ssc.start()
+    ssc.awaitTermination()
+  }
+}
+
+
+
+class StopContextThread(ssc: StreamingContext) extends Runnable {
+  def run {
+    ssc.stop(true, true)
+  }
+}
+
+
+class LatencyListener(ssc: StreamingContext) extends StreamingListener {
+
+  var metricMap: scala.collection.mutable.Map[String, Object] = _
+  var startTime = 0L
+  var startTime1 = 0L
+  var endTime = 0L
+  var endTime1 = 0L
+  var totalDelay = 0L
+  var hasStarted = false
+  var batchCount = 0
+  var totalRecords = 0L
+  val thread: Thread = new Thread(new StopContextThread(ssc))
+
+
+  def getMap(): scala.collection.mutable.Map[String, Object] = synchronized {
+    if (metricMap == null) metricMap = scala.collection.mutable.Map()
+    metricMap
+  }
+
+  def setMap(metricMap: scala.collection.mutable.Map[String, Object]) = synchronized {
+    this.metricMap = metricMap
+  }
+
+  /** Called when processing of a job of a batch has started. */
+  override def onOutputOperationStarted(outputOperationStarted: StreamingListenerOutputOperationStarted): Unit = {
+    println("job creation delay repoted in onOutputOperationStarted ==>"+outputOperationStarted.outputOperationInfo.batchTime+"==>"+outputOperationStarted.outputOperationInfo.id+"==>"+outputOperationStarted.outputOperationInfo.jobGenTime)
+  }
+
+  val batchSize =  Seconds(2).toString
+  val recordLimitPerThread = 1000
+  val loaderThreads = 10
+
+  val recordLimit = loaderThreads * recordLimitPerThread
+
+  override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {
+    val batchInfo = batchCompleted.batchInfo
+    val prevCount = totalRecords
+    var recordThisBatch = batchInfo.numRecords
+
+    println("job creation delay repoted in onBatchCompleted ==>" +batchInfo.batchTime+"==>"+batchInfo.batchJobSetCreationDelay )
+
+    if (!thread.isAlive) {
+      totalRecords += recordThisBatch
+      //      val imap = getMap
+      //      imap(batchInfo.batchTime.toString()) = "batchTime," + batchInfo.batchTime +
+      //        ", batch Count so far," + batchCount +
+      //        ", total Records so far," + totalRecords +
+      //        ", record This Batch," + recordThisBatch +
+      //        ", submission Time," + batchInfo.submissionTime +
+      //        ", processing Start Time," + batchInfo.processingStartTime.getOrElse(0L) +
+      //        ", processing End Time," + batchInfo.processingEndTime.getOrElse(0L) +
+      //        ", scheduling Delay," + batchInfo.schedulingDelay.getOrElse(0L) +
+      //        ", processing Delay," + batchInfo.processingDelay.getOrElse(0L)
+      //
+      //      setMap(imap)
+    }
+
+    if (totalRecords >= recordLimit) {
+      if (hasStarted && !thread.isAlive) {
+        //not receiving any data more, finish
+        endTime = System.currentTimeMillis()
+        endTime1 = batchInfo.processingEndTime.getOrElse(0L)
+        var warning = ""
+        val totalTime = (endTime - startTime).toDouble / 1000
+        //This is weighted avg of every batch process time. The weight is records processed int the batch
+        val avgLatency = totalDelay.toDouble / totalRecords
+        if (avgLatency > batchSize.toDouble)
+          warning = "WARNING:SPARK CLUSTER IN UNSTABLE STATE. TRY REDUCE INPUT SPEED"
+
+        val avgLatencyAdjust = avgLatency + batchSize.toDouble
+        val recordThroughput = totalRecords / totalTime
+
+        val imap = getMap
+
+        imap("Final Metric") = " Total Batch count," + batchCount +
+          ", startTime based on submissionTime," + startTime +
+          ", startTime based on System," + startTime1 +
+          ", endTime based on System," + endTime +
+          ", endTime based on processingEndTime," + endTime1 +
+          ", Total Records," + totalRecords +
+          // ", Total processing delay = " + totalDelay + " ms "+
+          ", Total Consumed time in sec," + totalTime +
+          ", Avg latency/batchInterval in ms," + avgLatencyAdjust +
+          ", Avg records/sec," + recordThroughput +
+          ", WARNING," + warning
+
+        imap.foreach { case (key, value) => println(key + "-->" + value) }
+
+        thread.start
+      }
+    } else if (!hasStarted) {
+      if (batchInfo.numRecords > 0) {
+        startTime = batchCompleted.batchInfo.submissionTime
+        startTime1 = System.currentTimeMillis()
+        hasStarted = true
+      }
+    }
+
+    if (hasStarted) {
+      //      println("This delay:"+batchCompleted.batchInfo.processingDelay+"ms")
+      batchCompleted.batchInfo.processingDelay match {
+        case Some(value) => totalDelay += value * recordThisBatch
+        case None => //Nothing
+      }
+      batchCount = batchCount + 1
+    }
+  }
+}
+
 object DirectKafkaStreamSuite {
   val collectedData = new ConcurrentLinkedQueue[String]()
   @volatile var total = -1L

streaming/src/main/scala/org/apache/spark/streaming/DStreamGraph.scala

@@ -114,8 +114,10 @@ final private[streaming] class DStreamGraph extends Serializable with Logging {
     logDebug("Generating jobs for time " + time)
     val jobs = this.synchronized {
       outputStreams.flatMap { outputStream =>
+        val genStartTime=System.currentTimeMillis()
         val jobOption = outputStream.generateJob(time)
         jobOption.foreach(_.setCallSite(outputStream.creationSite))
+        jobOption.foreach(_.setGenDelay(System.currentTimeMillis()-genStartTime))
         jobOption
       }
     }

streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListener.scala

@@ -241,4 +241,5 @@ private[streaming] case class JavaOutputOperationInfo(
     description: String,
     startTime: Long,
     endTime: Long,
+    jobGenTime:  Long,
     failureReason: String)

streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListenerWrapper.scala

@@ -58,6 +58,7 @@ private[streaming] class JavaStreamingListenerWrapper(javaStreamingListener: Jav
       outputOperationInfo.description: String,
       outputOperationInfo.startTime.getOrElse(-1),
       outputOperationInfo.endTime.getOrElse(-1),
+      outputOperationInfo.jobGenTime.getOrElse(-1),
       outputOperationInfo.failureReason.orNull
     )
   }

streaming/src/main/scala/org/apache/spark/streaming/scheduler/Job.scala

@@ -34,6 +34,7 @@ class Job(val time: Time, func: () => _) {
   private var _callSite: CallSite = null
   private var _startTime: Option[Long] = None
   private var _endTime: Option[Long] = None
+  private var _jobGenTime: Option[Long] = None
 
   def run() {
     _result = Try(func())
@@ -85,6 +86,14 @@ class Job(val time: Time, func: () => _) {
     _startTime = Some(startTime)
   }
 
+  def setGenDelay(jobGenTime: Long): Unit = {
+    _jobGenTime = Some(jobGenTime)
+  }
+
+  def getGenDelay(): Option[Long]  = {
+    _jobGenTime
+  }
+
   def setEndTime(endTime: Long): Unit = {
     _endTime = Some(endTime)
   }
@@ -96,7 +105,7 @@ class Job(val time: Time, func: () => _) {
       None
     }
     OutputOperationInfo(
-      time, outputOpId, callSite.shortForm, callSite.longForm, _startTime, _endTime, failureReason)
+      time, outputOpId, callSite.shortForm, callSite.longForm, _startTime, _endTime, _jobGenTime, failureReason)
   }
 
   override def toString: String = id

streaming/src/main/scala/org/apache/spark/streaming/scheduler/OutputOperationInfo.scala

@@ -39,6 +39,7 @@ case class OutputOperationInfo(
     description: String,
     startTime: Option[Long],
     endTime: Option[Long],
+    jobGenTime:  Option[Long],
     failureReason: Option[String]) {
 
   /**

streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala

@@ -116,6 +116,7 @@ private[ui] case class OutputOperationUIData(
     description: String,
     startTime: Option[Long],
     endTime: Option[Long],
+    jobGenTime: Option[Long],
     failureReason: Option[String]) {
 
   def duration: Option[Long] = for (s <- startTime; e <- endTime) yield e - s
@@ -130,6 +131,7 @@ private[ui] object OutputOperationUIData {
       outputOperationInfo.description,
       outputOperationInfo.startTime,
       outputOperationInfo.endTime,
+      outputOperationInfo.jobGenTime,
       outputOperationInfo.failureReason
     )
   }

streaming/src/test/java/org/apache/spark/streaming/api/java/JavaStreamingListenerWrapperSuite.scala

@@ -84,6 +84,7 @@ class JavaStreamingListenerWrapperSuite extends SparkFunSuite {
           description = "operation1",
           startTime = None,
           endTime = None,
+          jobGenTime = None,
           failureReason = None),
         1 -> OutputOperationInfo(
           batchTime = Time(1000L),
@@ -92,6 +93,7 @@ class JavaStreamingListenerWrapperSuite extends SparkFunSuite {
           description = "operation2",
           startTime = None,
           endTime = None,
+          jobGenTime = None,
           failureReason = None))
     ))
     listenerWrapper.onBatchSubmitted(batchSubmitted)
@@ -119,6 +121,7 @@ class JavaStreamingListenerWrapperSuite extends SparkFunSuite {
           description = "operation1",
           startTime = Some(1003L),
           endTime = None,
+          jobGenTime = None,
           failureReason = None),
         1 -> OutputOperationInfo(
           batchTime = Time(1000L),
@@ -127,6 +130,7 @@ class JavaStreamingListenerWrapperSuite extends SparkFunSuite {
           description = "operation2",
           startTime = Some(1005L),
           endTime = None,
+          jobGenTime = None,
           failureReason = None))
     ))
     listenerWrapper.onBatchStarted(batchStarted)
@@ -154,6 +158,7 @@ class JavaStreamingListenerWrapperSuite extends SparkFunSuite {
           description = "operation1",
           startTime = Some(1003L),
           endTime = Some(1004L),
+          jobGenTime = None,
           failureReason = None),
         1 -> OutputOperationInfo(
           batchTime = Time(1000L),
@@ -162,6 +167,7 @@ class JavaStreamingListenerWrapperSuite extends SparkFunSuite {
           description = "operation2",
           startTime = Some(1005L),
           endTime = Some(1010L),
+          jobGenTime = None,
           failureReason = None))
     ))
     listenerWrapper.onBatchCompleted(batchCompleted)
@@ -174,6 +180,7 @@ class JavaStreamingListenerWrapperSuite extends SparkFunSuite {
       description = "operation1",
       startTime = Some(1003L),
       endTime = None,
+      jobGenTime = None,
       failureReason = None
     ))
     listenerWrapper.onOutputOperationStarted(outputOperationStarted)
@@ -187,6 +194,7 @@ class JavaStreamingListenerWrapperSuite extends SparkFunSuite {
       description = "operation1",
       startTime = Some(1003L),
       endTime = Some(1004L),
+      jobGenTime = None,
       failureReason = None
     ))
     listenerWrapper.onOutputOperationCompleted(outputOperationCompleted)
@@ -243,6 +251,7 @@ class JavaStreamingListenerWrapperSuite extends SparkFunSuite {
     assert(javaOutputOperationInfo.description === outputOperationInfo.description)
     assert(javaOutputOperationInfo.startTime === outputOperationInfo.startTime.getOrElse(-1))
     assert(javaOutputOperationInfo.endTime === outputOperationInfo.endTime.getOrElse(-1))
+    assert(javaOutputOperationInfo.jobGenTime === outputOperationInfo.jobGenTime.getOrElse(-1))
     assert(javaOutputOperationInfo.failureReason === outputOperationInfo.failureReason.orNull)
   }
 }