Rename: scheduledLocations -> scheduledExecutors; locations -> executors

zsxwing · zsxwing · commit 715ef9caf7ce · 2015-07-23T16:11:18.000+08:00
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverSchedulingPolicy.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverSchedulingPolicy.scala
@@ -42,55 +42,54 @@ private[streaming] class ReceiverSchedulingPolicy {
     }
 
     val hostToExecutors = executors.groupBy(_.split(":")(0))
-    val locations = new Array[mutable.ArrayBuffer[String]](receivers.length)
+    val scheduledExecutors = Array.fill(receivers.length)(new mutable.ArrayBuffer[String])
     val numReceiversOnExecutor = mutable.HashMap[String, Int]()
     // Set the initial value to 0
     executors.foreach(numReceiversOnExecutor(_) = 0)
 
     // Firstly, we need to respect "preferredLocation". So if a receiver has "preferredLocation",
-    // we need to make sure the "preferredLocation" is in the candidate location list.
+    // we need to make sure the "preferredLocation" is in the candidate scheduled executor list.
     for (i <- 0 until receivers.length) {
-      locations(i) = new mutable.ArrayBuffer[String]()
       // Note: preferredLocation is host but executors are host:port
       receivers(i).preferredLocation.foreach { host =>
         hostToExecutors.get(host) match {
           case Some(executorsOnHost) =>
             // preferredLocation is a known host. Select an executor that has the least receivers in
             // this host
-            val scheduledLocation =
+            val leastScheduledExecutor =
               executorsOnHost.minBy(executor => numReceiversOnExecutor(executor))
-            locations(i) += scheduledLocation
-            numReceiversOnExecutor(scheduledLocation) =
-              numReceiversOnExecutor(scheduledLocation) + 1
+            scheduledExecutors(i) += leastScheduledExecutor
+            numReceiversOnExecutor(leastScheduledExecutor) =
+              numReceiversOnExecutor(leastScheduledExecutor) + 1
           case None =>
             // preferredLocation is an unknown host.
             // Note: There are two cases:
             // 1. This executor is not up. But it may be up later.
             // 2. This executor is dead, or it's not a host in the cluster.
-            // Currently, simply add host to the scheduled locations
-            locations(i) += host
+            // Currently, simply add host to the scheduled executors.
+            scheduledExecutors(i) += host
         }
       }
     }
 
     // For those receivers that don't have preferredLocation, make sure we assign at least one
     // executor to them.
-    for (scheduledLocations <- locations.filter(_.isEmpty)) {
+    for (scheduledExecutorsForOneReceiver <- scheduledExecutors.filter(_.isEmpty)) {
       // Select the executor that has the least receivers
-      val (executor, numReceivers) = numReceiversOnExecutor.minBy(_._2)
-      scheduledLocations += executor
-      numReceiversOnExecutor(executor) = numReceivers + 1
+      val (leastScheduledExecutor, numReceivers) = numReceiversOnExecutor.minBy(_._2)
+      scheduledExecutorsForOneReceiver += leastScheduledExecutor
+      numReceiversOnExecutor(leastScheduledExecutor) = numReceivers + 1
     }
 
     // Assign idle executors to receivers that have less executors
     val idleExecutors = numReceiversOnExecutor.filter(_._2 == 0).map(_._1)
     for (executor <- idleExecutors) {
-      // Assign an idle executor to the receiver that has least locations.
-      val scheduledLocations = locations.minBy(_.size)
-      scheduledLocations += executor
+      // Assign an idle executor to the receiver that has least candidate executors.
+      val leastScheduledExecutors = scheduledExecutors.minBy(_.size)
+      leastScheduledExecutors += executor
     }
 
-    receivers.map(_.streamId).zip(locations).toMap
+    receivers.map(_.streamId).zip(scheduledExecutors).toMap
   }
 
   /**
@@ -131,31 +130,31 @@ private[streaming] class ReceiverSchedulingPolicy {
     }
 
     // Always try to schedule to the preferred locations
-    val locations = mutable.Set[String]()
-    locations ++= preferredLocation
+    val scheduledExecutors = mutable.Set[String]()
+    scheduledExecutors ++= preferredLocation
 
     val executorWeights = receiverTrackingInfoMap.values.flatMap { receiverTrackingInfo =>
       receiverTrackingInfo.state match {
         case ReceiverState.INACTIVE => Nil
         case ReceiverState.SCHEDULED =>
-          val scheduledLocations = receiverTrackingInfo.scheduledLocations.get
+          val scheduledExecutors = receiverTrackingInfo.scheduledExecutors.get
           // The probability that a scheduled receiver will run in an executor is
           // 1.0 / scheduledLocations.size
-          scheduledLocations.map(location => location -> (1.0 / scheduledLocations.size))
-        case ReceiverState.ACTIVE => Seq(receiverTrackingInfo.runningLocation.get -> 1.0)
+          scheduledExecutors.map(location => location -> (1.0 / scheduledExecutors.size))
+        case ReceiverState.ACTIVE => Seq(receiverTrackingInfo.runningExecutor.get -> 1.0)
       }
     }.groupBy(_._1).mapValues(_.map(_._2).sum) // Sum weights for each executor
 
     val idleExecutors = (executors.toSet -- executorWeights.keys).toSeq
     if (idleExecutors.size >= 3) {
       // If there are more than 3 idle executors, return all of them
-      locations ++= idleExecutors
+      scheduledExecutors ++= idleExecutors
     } else {
       // If there are less than 3 idle executors, return 3 best options
-      locations ++= idleExecutors
+      scheduledExecutors ++= idleExecutors
       val sortedExecutors = executorWeights.toSeq.sortBy(_._2).map(_._1)
-      locations ++= (idleExecutors ++ sortedExecutors).take(3)
+      scheduledExecutors ++= (idleExecutors ++ sortedExecutors).take(3)
     }
-    locations.toSeq
+    scheduledExecutors.toSeq
   }
 }
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
@@ -249,8 +249,8 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
       val receiverTrackingInfo = ReceiverTrackingInfo(
         streamId,
         ReceiverState.ACTIVE,
-        scheduledLocations = None,
-        runningLocation = Some(hostPort),
+        scheduledExecutors = None,
+        runningExecutor = Some(hostPort),
         name = Some(name),
         endpoint = Some(receiverEndpoint))
       receiverTrackingInfos.put(streamId, receiverTrackingInfo)
@@ -321,23 +321,23 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
 
   private def scheduleReceiver(receiverId: Int): Seq[String] = {
     val preferredLocation = receiverPreferredLocations.getOrElse(receiverId, None)
-    val scheduledLocations = schedulingPolicy.rescheduleReceiver(
+    val scheduledExecutors = schedulingPolicy.rescheduleReceiver(
       receiverId, preferredLocation, receiverTrackingInfos, getExecutors)
-    updateReceiverScheduledLocations(receiverId, scheduledLocations)
-    scheduledLocations
+    updateReceiverScheduledExecutors(receiverId, scheduledExecutors)
+    scheduledExecutors
   }
 
-  private def updateReceiverScheduledLocations(
-      receiverId: Int, scheduledLocations: Seq[String]): Unit = {
+  private def updateReceiverScheduledExecutors(
+      receiverId: Int, scheduledExecutors: Seq[String]): Unit = {
     val newReceiverTrackingInfo = receiverTrackingInfos.get(receiverId) match {
       case Some(oldInfo) =>
         oldInfo.copy(state = ReceiverState.SCHEDULED,
-          scheduledLocations = Some(scheduledLocations))
+          scheduledExecutors = Some(scheduledExecutors))
       case None =>
         ReceiverTrackingInfo(
           receiverId,
           ReceiverState.SCHEDULED,
-          Some(scheduledLocations),
+          Some(scheduledExecutors),
           None)
     }
     receiverTrackingInfos.put(receiverId, newReceiverTrackingInfo)
@@ -412,21 +412,21 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
     override def receive: PartialFunction[Any, Unit] = {
       // Local messages
       case StartAllReceivers(receivers) =>
-        val scheduledLocations = schedulingPolicy.scheduleReceivers(receivers, getExecutors)
+        val scheduledExecutors = schedulingPolicy.scheduleReceivers(receivers, getExecutors)
         for (receiver <- receivers) {
-          val locations = scheduledLocations(receiver.streamId)
-          updateReceiverScheduledLocations(receiver.streamId, locations)
+          val executors = scheduledExecutors(receiver.streamId)
+          updateReceiverScheduledExecutors(receiver.streamId, executors)
           receiverPreferredLocations(receiver.streamId) = receiver.preferredLocation
-          startReceiver(receiver, locations)
+          startReceiver(receiver, executors)
         }
       case RestartReceiver(receiver) =>
-        val scheduledLocations = schedulingPolicy.rescheduleReceiver(
+        val scheduledExecutors = schedulingPolicy.rescheduleReceiver(
           receiver.streamId,
           receiver.preferredLocation,
           receiverTrackingInfos,
           getExecutors)
-        updateReceiverScheduledLocations(receiver.streamId, scheduledLocations)
-        startReceiver(receiver, scheduledLocations)
+        updateReceiverScheduledExecutors(receiver.streamId, scheduledExecutors)
+        startReceiver(receiver, scheduledExecutors)
       case c: CleanupOldBlocks =>
         receiverTrackingInfos.values.flatMap(_.endpoint).foreach(_.send(c))
       case UpdateReceiverRateLimit(streamUID, newRate) =>
@@ -459,9 +459,9 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
     }
 
     /**
-     * Start a receiver along with its scheduled locations
+     * Start a receiver along with its scheduled executors
      */
-    private def startReceiver(receiver: Receiver[_], scheduledLocations: Seq[String]): Unit = {
+    private def startReceiver(receiver: Receiver[_], scheduledExecutors: Seq[String]): Unit = {
       val receiverId = receiver.streamId
       if (!isTrackerStarted) {
         onReceiverJobFinish(receiverId)
@@ -475,12 +475,12 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
       // Function to start the receiver on the worker node
       val startReceiverFunc = new StartReceiverFunc(checkpointDirOption, serializableHadoopConf)
 
-      // Create the RDD using the scheduledLocations to run the receiver in a Spark job
+      // Create the RDD using the scheduledExecutors to run the receiver in a Spark job
       val receiverRDD: RDD[Receiver[_]] =
-        if (scheduledLocations.isEmpty) {
+        if (scheduledExecutors.isEmpty) {
           ssc.sc.makeRDD(Seq(receiver), 1)
         } else {
-          ssc.sc.makeRDD(Seq(receiver -> scheduledLocations))
+          ssc.sc.makeRDD(Seq(receiver -> scheduledExecutors))
         }
       receiverRDD.setName(s"Receiver $receiverId")
       val future = ssc.sparkContext.submitJob[Receiver[_], Unit, Unit](
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTrackingInfo.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTrackingInfo.scala
@@ -28,17 +28,17 @@ private[streaming] case class ReceiverErrorInfo(
  *
  * @param receiverId the unique receiver id
  * @param state the current Receiver state
- * @param scheduledLocations the scheduled locations provided by ReceiverSchedulingPolicy
- * @param runningLocation the running location if the receiver is active
+ * @param scheduledExecutors the scheduled executors provided by ReceiverSchedulingPolicy
+ * @param runningExecutor the running executor if the receiver is active
  * @param name the receiver name
  * @param endpoint the receiver endpoint. It can be used to send messages to the receiver
  * @param errorInfo the receiver error information if it fails
  */
 private[streaming] case class ReceiverTrackingInfo(
     receiverId: Int,
     state: ReceiverState,
-    scheduledLocations: Option[Seq[String]],
-    runningLocation: Option[String],
+    scheduledExecutors: Option[Seq[String]],
+    runningExecutor: Option[String],
     name: Option[String] = None,
     endpoint: Option[RpcEndpointRef] = None,
     errorInfo: Option[ReceiverErrorInfo] = None) {
@@ -47,7 +47,7 @@ private[streaming] case class ReceiverTrackingInfo(
     receiverId,
     name.getOrElse(""),
     state == ReceiverState.ACTIVE,
-    location = runningLocation.getOrElse(""),
+    location = runningExecutor.getOrElse(""),
     lastErrorMessage = errorInfo.map(_.lastErrorMessage).getOrElse(""),
     lastError = errorInfo.map(_.lastError).getOrElse(""),
     lastErrorTime = errorInfo.map(_.lastErrorTime).getOrElse(-1L)
diff --git a/streaming/src/test/scala/org/apache/spark/streaming/scheduler/ReceiverSchedulingPolicySuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/scheduler/ReceiverSchedulingPolicySuite.scala
@@ -28,27 +28,27 @@ class ReceiverSchedulingPolicySuite extends SparkFunSuite {
   val receiverSchedulingPolicy = new ReceiverSchedulingPolicy
 
   test("rescheduleReceiver: empty executors") {
-    val scheduledLocations =
+    val scheduledExecutors =
       receiverSchedulingPolicy.rescheduleReceiver(0, None, Map.empty, executors = Seq.empty)
-    assert(scheduledLocations === Seq.empty)
+    assert(scheduledExecutors === Seq.empty)
   }
 
   test("rescheduleReceiver: receiver preferredLocation") {
     val receiverTrackingInfoMap = Map(
       0 -> ReceiverTrackingInfo(0, ReceiverState.INACTIVE, None, None))
-    val scheduledLocations = receiverSchedulingPolicy.rescheduleReceiver(
+    val scheduledExecutors = receiverSchedulingPolicy.rescheduleReceiver(
       0, Some("host1"), receiverTrackingInfoMap, executors = Seq("host2"))
-    assert(scheduledLocations.toSet === Set("host1", "host2"))
+    assert(scheduledExecutors.toSet === Set("host1", "host2"))
   }
 
   test("rescheduleReceiver: return all idle executors if more than 3 idle executors") {
     val executors = Seq("host1", "host2", "host3", "host4", "host5")
     // host3 is idle
     val receiverTrackingInfoMap = Map(
       0 -> ReceiverTrackingInfo(0, ReceiverState.ACTIVE, None, Some("host1")))
-    val scheduledLocations = receiverSchedulingPolicy.rescheduleReceiver(
+    val scheduledExecutors = receiverSchedulingPolicy.rescheduleReceiver(
       1, None, receiverTrackingInfoMap, executors)
-    assert(scheduledLocations.toSet === Set("host2", "host3", "host4", "host5"))
+    assert(scheduledExecutors.toSet === Set("host2", "host3", "host4", "host5"))
   }
 
   test("rescheduleReceiver: return 3 best options if less than 3 idle executors") {
@@ -59,21 +59,21 @@ class ReceiverSchedulingPolicySuite extends SparkFunSuite {
       0 -> ReceiverTrackingInfo(0, ReceiverState.ACTIVE, None, Some("host1")),
       1 -> ReceiverTrackingInfo(1, ReceiverState.SCHEDULED, Some(Seq("host2", "host3")), None),
       2 -> ReceiverTrackingInfo(1, ReceiverState.SCHEDULED, Some(Seq("host1", "host3")), None))
-    val scheduledLocations = receiverSchedulingPolicy.rescheduleReceiver(
+    val scheduledExecutors = receiverSchedulingPolicy.rescheduleReceiver(
       3, None, receiverTrackingInfoMap, executors)
-    assert(scheduledLocations.toSet === Set("host2", "host4", "host5"))
+    assert(scheduledExecutors.toSet === Set("host2", "host4", "host5"))
   }
 
   test("scheduleReceivers: " +
     "schedule receivers evenly when there are more receivers than executors") {
     val receivers = (0 until 6).map(new DummyReceiver(_))
     val executors = (10000 until 10003).map(port => s"localhost:${port}")
-    val scheduledLocations = receiverSchedulingPolicy.scheduleReceivers(receivers, executors)
+    val scheduledExecutors = receiverSchedulingPolicy.scheduleReceivers(receivers, executors)
     val numReceiversOnExecutor = mutable.HashMap[String, Int]()
-    // There should be 2 receivers running on each executor and each receiver has one location
-    scheduledLocations.foreach { case (receiverId, locations) =>
-      assert(locations.size == 1)
-      numReceiversOnExecutor(locations(0)) = numReceiversOnExecutor.getOrElse(locations(0), 0) + 1
+    // There should be 2 receivers running on each executor and each receiver has one executor
+    scheduledExecutors.foreach { case (receiverId, executors) =>
+      assert(executors.size == 1)
+      numReceiversOnExecutor(executors(0)) = numReceiversOnExecutor.getOrElse(executors(0), 0) + 1
     }
     assert(numReceiversOnExecutor === executors.map(_ -> 2).toMap)
   }
@@ -83,12 +83,12 @@ class ReceiverSchedulingPolicySuite extends SparkFunSuite {
     "schedule receivers evenly when there are more executors than receivers") {
     val receivers = (0 until 3).map(new DummyReceiver(_))
     val executors = (10000 until 10006).map(port => s"localhost:${port}")
-    val scheduledLocations = receiverSchedulingPolicy.scheduleReceivers(receivers, executors)
+    val scheduledExecutors = receiverSchedulingPolicy.scheduleReceivers(receivers, executors)
     val numReceiversOnExecutor = mutable.HashMap[String, Int]()
-    // There should be 1 receiver running on each executor and each receiver has two locations
-    scheduledLocations.foreach { case (receiverId, locations) =>
-      assert(locations.size == 2)
-      locations.foreach { l =>
+    // There should be 1 receiver running on each executor and each receiver has two executors
+    scheduledExecutors.foreach { case (receiverId, executors) =>
+      assert(executors.size == 2)
+      executors.foreach { l =>
         numReceiversOnExecutor(l) = numReceiversOnExecutor.getOrElse(l, 0) + 1
       }
     }
@@ -100,33 +100,33 @@ class ReceiverSchedulingPolicySuite extends SparkFunSuite {
       (3 until 6).map(new DummyReceiver(_, Some("localhost")))
     val executors = (10000 until 10003).map(port => s"localhost:${port}") ++
       (10003 until 10006).map(port => s"localhost2:${port}")
-    val scheduledLocations = receiverSchedulingPolicy.scheduleReceivers(receivers, executors)
+    val scheduledExecutors = receiverSchedulingPolicy.scheduleReceivers(receivers, executors)
     val numReceiversOnExecutor = mutable.HashMap[String, Int]()
-    // There should be 1 receiver running on each executor and each receiver has 1 location
-    scheduledLocations.foreach { case (receiverId, locations) =>
-      assert(locations.size == 1)
-      locations.foreach { l =>
+    // There should be 1 receiver running on each executor and each receiver has 1 executor
+    scheduledExecutors.foreach { case (receiverId, executors) =>
+      assert(executors.size == 1)
+      executors.foreach { l =>
         numReceiversOnExecutor(l) = numReceiversOnExecutor.getOrElse(l, 0) + 1
       }
     }
     assert(numReceiversOnExecutor === executors.map(_ -> 1).toMap)
     // Make sure we schedule the receivers to their preferredLocations
-    val locationsForReceiversWithPreferredLocation =
-      scheduledLocations.filter { case (receiverId, locations) => receiverId >= 3 }.flatMap(_._2)
-    // We can simply check the location set because we only know each receiver only has 1 location
-    assert(locationsForReceiversWithPreferredLocation.toSet ===
+    val executorsForReceiversWithPreferredLocation =
+      scheduledExecutors.filter { case (receiverId, executors) => receiverId >= 3 }.flatMap(_._2)
+    // We can simply check the executor set because we only know each receiver only has 1 executor
+    assert(executorsForReceiversWithPreferredLocation.toSet ===
       (10000 until 10003).map(port => s"localhost:${port}").toSet)
   }
 
   test("scheduleReceivers: return empty if no receiver") {
     assert(receiverSchedulingPolicy.scheduleReceivers(Seq.empty, Seq("localhost:10000")).isEmpty)
   }
 
-  test("scheduleReceivers: return empty locations if no executors") {
+  test("scheduleReceivers: return empty scheduled executors if no executors") {
     val receivers = (0 until 3).map(new DummyReceiver(_))
-    val scheduledLocations = receiverSchedulingPolicy.scheduleReceivers(receivers, Seq.empty)
-    scheduledLocations.foreach { case (receiverId, locations) =>
-      assert(locations.isEmpty)
+    val scheduledExecutors = receiverSchedulingPolicy.scheduleReceivers(receivers, Seq.empty)
+    scheduledExecutors.foreach { case (receiverId, executors) =>
+      assert(executors.isEmpty)
     }
   }
 }
