Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/9181#discussion_r43093391
--- Diff:
streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverSchedulingPolicy.scala
---
@@ -163,40 +169,59 @@ private[streaming] class ReceiverSchedulingPolicy {
receiverId: Int,
preferredLocation: Option[String],
receiverTrackingInfoMap: Map[Int, ReceiverTrackingInfo],
- executors: Seq[String]): Seq[String] = {
+ executors: Seq[ExecutorCacheTaskLocation]): Seq[TaskLocation] = {
if (executors.isEmpty) {
return Seq.empty
}
// Always try to schedule to the preferred locations
- val scheduledExecutors = mutable.Set[String]()
- scheduledExecutors ++= preferredLocation
-
- val executorWeights = receiverTrackingInfoMap.values.flatMap {
receiverTrackingInfo =>
- receiverTrackingInfo.state match {
- case ReceiverState.INACTIVE => Nil
- case ReceiverState.SCHEDULED =>
- val scheduledExecutors =
receiverTrackingInfo.scheduledExecutors.get
- // The probability that a scheduled receiver will run in an
executor is
- // 1.0 / scheduledLocations.size
- 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 scheduledLocations = mutable.Set[TaskLocation]()
+ // Note: preferredLocation could be `HDFSCacheTaskLocation`, so use
`TaskLocation.apply` to
+ // handle this case
+ scheduledLocations ++= preferredLocation.map(TaskLocation(_))
+
+ val executorWeights: Map[ExecutorCacheTaskLocation, Double] = {
+
receiverTrackingInfoMap.values.flatMap(convertReceiverTrackingInfoToExecutorWeights)
+ .groupBy(_._1).mapValues(_.map(_._2).sum) // Sum weights for each
executor
+ }
val idleExecutors = executors.toSet -- executorWeights.keys
if (idleExecutors.nonEmpty) {
- scheduledExecutors ++= idleExecutors
+ scheduledLocations ++= idleExecutors
} else {
// There is no idle executor. So select all executors that have the
minimum weight.
val sortedExecutors = executorWeights.toSeq.sortBy(_._2)
if (sortedExecutors.nonEmpty) {
val minWeight = sortedExecutors(0)._2
- scheduledExecutors ++= sortedExecutors.takeWhile(_._2 ==
minWeight).map(_._1)
+ scheduledLocations ++= sortedExecutors.takeWhile(_._2 ==
minWeight).map(_._1)
} else {
// This should not happen since "executors" is not empty
}
}
- scheduledExecutors.toSeq
+ scheduledLocations.toSeq
+ }
+
+ /**
+ * This method tries to convert a receiver tracking info to executor
weights. Every executor will
+ * be assigned to a weight according to the receivers running or
scheduling on it:
+ *
+ * - If a receiver is running on an executor, it contributes 1.0 to the
executor's weight.
+ * - If a receiver is scheduled to an executor but has not yet run, it
contributes
+ * `1.0 / #candidate_executors_of_this_receiver` to the executor's
weight.
+ */
+ private def convertReceiverTrackingInfoToExecutorWeights(
+ receiverTrackingInfo: ReceiverTrackingInfo):
Seq[(ExecutorCacheTaskLocation, Double)] = {
+ receiverTrackingInfo.state match {
+ case ReceiverState.INACTIVE => Nil
+ case ReceiverState.SCHEDULED =>
+ val scheduledLocations =
receiverTrackingInfo.scheduledLocations.get
+ // The probability that a scheduled receiver will run in an
executor is
+ // 1.0 / scheduledLocations.size
+
scheduledLocations.filter(_.isInstanceOf[ExecutorCacheTaskLocation]).
+ map { location =>
--- End diff --
nit:shouldnt this map be in the line above? I think `.map { loc =>` would
fit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
