|
| 1 | +package org.apache.spark.streaming.ui |
| 2 | + |
| 3 | +import org.apache.spark.streaming.{Time, StreamingContext} |
| 4 | +import org.apache.spark.streaming.scheduler._ |
| 5 | +import scala.collection.mutable.{Queue, HashMap} |
| 6 | +import org.apache.spark.streaming.scheduler.StreamingListenerReceiverStarted |
| 7 | +import org.apache.spark.streaming.scheduler.StreamingListenerBatchStarted |
| 8 | +import org.apache.spark.streaming.scheduler.BatchInfo |
| 9 | +import org.apache.spark.streaming.scheduler.ReceiverInfo |
| 10 | +import org.apache.spark.streaming.scheduler.StreamingListenerBatchSubmitted |
| 11 | +import org.apache.spark.util.Distribution |
| 12 | + |
| 13 | + |
| 14 | +private[ui] class StreamingProgressListener(ssc: StreamingContext) extends StreamingListener { |
| 15 | + |
| 16 | + private val waitingBatchInfos = new HashMap[Time, BatchInfo] |
| 17 | + private val runningBatchInfos = new HashMap[Time, BatchInfo] |
| 18 | + private val completedaBatchInfos = new Queue[BatchInfo] |
| 19 | + private val batchInfoLimit = ssc.conf.getInt("spark.steaming.ui.maxBatches", 100) |
| 20 | + private var totalCompletedBatches = 0L |
| 21 | + private val receiverInfos = new HashMap[Int, ReceiverInfo] |
| 22 | + |
| 23 | + val batchDuration = ssc.graph.batchDuration.milliseconds |
| 24 | + |
| 25 | + override def onReceiverStarted(receiverStarted: StreamingListenerReceiverStarted) = { |
| 26 | + synchronized { |
| 27 | + receiverInfos.put(receiverStarted.receiverInfo.streamId, receiverStarted.receiverInfo) |
| 28 | + } |
| 29 | + } |
| 30 | + |
| 31 | + override def onBatchSubmitted(batchSubmitted: StreamingListenerBatchSubmitted) = synchronized { |
| 32 | + runningBatchInfos(batchSubmitted.batchInfo.batchTime) = batchSubmitted.batchInfo |
| 33 | + } |
| 34 | + |
| 35 | + override def onBatchStarted(batchStarted: StreamingListenerBatchStarted) = synchronized { |
| 36 | + runningBatchInfos(batchStarted.batchInfo.batchTime) = batchStarted.batchInfo |
| 37 | + waitingBatchInfos.remove(batchStarted.batchInfo.batchTime) |
| 38 | + } |
| 39 | + |
| 40 | + override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted) = synchronized { |
| 41 | + waitingBatchInfos.remove(batchCompleted.batchInfo.batchTime) |
| 42 | + runningBatchInfos.remove(batchCompleted.batchInfo.batchTime) |
| 43 | + completedaBatchInfos.enqueue(batchCompleted.batchInfo) |
| 44 | + if (completedaBatchInfos.size > batchInfoLimit) completedaBatchInfos.dequeue() |
| 45 | + totalCompletedBatches += 1L |
| 46 | + } |
| 47 | + |
| 48 | + def numNetworkReceivers = synchronized { |
| 49 | + ssc.graph.getNetworkInputStreams().size |
| 50 | + } |
| 51 | + |
| 52 | + def numTotalCompletedBatches: Long = synchronized { |
| 53 | + totalCompletedBatches |
| 54 | + } |
| 55 | + |
| 56 | + def numUnprocessedBatches: Long = synchronized { |
| 57 | + waitingBatchInfos.size + runningBatchInfos.size |
| 58 | + } |
| 59 | + |
| 60 | + def waitingBatches: Seq[BatchInfo] = synchronized { |
| 61 | + waitingBatchInfos.values.toSeq |
| 62 | + } |
| 63 | + |
| 64 | + def runningBatches: Seq[BatchInfo] = synchronized { |
| 65 | + runningBatchInfos.values.toSeq |
| 66 | + } |
| 67 | + |
| 68 | + def completedBatches: Seq[BatchInfo] = synchronized { |
| 69 | + completedaBatchInfos.toSeq |
| 70 | + } |
| 71 | + |
| 72 | + def processingDelayDistribution: Option[Distribution] = synchronized { |
| 73 | + extractDistribution(_.processingDelay) |
| 74 | + } |
| 75 | + |
| 76 | + def schedulingDelayDistribution: Option[Distribution] = synchronized { |
| 77 | + extractDistribution(_.schedulingDelay) |
| 78 | + } |
| 79 | + |
| 80 | + def totalDelayDistribution: Option[Distribution] = synchronized { |
| 81 | + extractDistribution(_.totalDelay) |
| 82 | + } |
| 83 | + |
| 84 | + def receivedRecordsDistributions: Map[Int, Option[Distribution]] = synchronized { |
| 85 | + val latestBatchInfos = allBatches.reverse.take(batchInfoLimit) |
| 86 | + val latestBlockInfos = latestBatchInfos.map(_.receivedBlockInfo) |
| 87 | + (0 until numNetworkReceivers).map { receiverId => |
| 88 | + val blockInfoOfParticularReceiver = latestBlockInfos.map { batchInfo => |
| 89 | + batchInfo.get(receiverId).getOrElse(Array.empty) |
| 90 | + } |
| 91 | + val recordsOfParticularReceiver = blockInfoOfParticularReceiver.map { blockInfo => |
| 92 | + // calculate records per second for each batch |
| 93 | + blockInfo.map(_.numRecords).sum.toDouble * 1000 / batchDuration |
| 94 | + } |
| 95 | + val distributionOption = Distribution(recordsOfParticularReceiver) |
| 96 | + (receiverId, distributionOption) |
| 97 | + }.toMap |
| 98 | + } |
| 99 | + |
| 100 | + def lastReceivedBatchRecords: Map[Int, Long] = { |
| 101 | + val lastReceivedBlockInfoOption = lastReceivedBatch.map(_.receivedBlockInfo) |
| 102 | + lastReceivedBlockInfoOption.map { lastReceivedBlockInfo => |
| 103 | + (0 until numNetworkReceivers).map { receiverId => |
| 104 | + (receiverId, lastReceivedBlockInfo(receiverId).map(_.numRecords).sum) |
| 105 | + }.toMap |
| 106 | + }.getOrElse { |
| 107 | + (0 until numNetworkReceivers).map(receiverId => (receiverId, 0L)).toMap |
| 108 | + } |
| 109 | + } |
| 110 | + |
| 111 | + def receiverInfo(receiverId: Int): Option[ReceiverInfo] = { |
| 112 | + receiverInfos.get(receiverId) |
| 113 | + } |
| 114 | + |
| 115 | + def lastCompletedBatch: Option[BatchInfo] = { |
| 116 | + completedaBatchInfos.sortBy(_.batchTime)(Time.ordering).lastOption |
| 117 | + } |
| 118 | + |
| 119 | + def lastReceivedBatch: Option[BatchInfo] = { |
| 120 | + allBatches.lastOption |
| 121 | + } |
| 122 | + |
| 123 | + private def allBatches: Seq[BatchInfo] = synchronized { |
| 124 | + (waitingBatchInfos.values.toSeq ++ |
| 125 | + runningBatchInfos.values.toSeq ++ completedaBatchInfos).sortBy(_.batchTime)(Time.ordering) |
| 126 | + } |
| 127 | + |
| 128 | + private def extractDistribution(getMetric: BatchInfo => Option[Long]): Option[Distribution] = { |
| 129 | + Distribution(completedaBatchInfos.flatMap(getMetric(_)).map(_.toDouble)) |
| 130 | + } |
| 131 | +} |
0 commit comments