blob: e492279b4ec6444dfc793f0a0a20309c60b8c399 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
package spark.scheduler
import spark._
private[spark] class ResultTask[T, U](
stageId: Int,
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
val partition: Int,
@transient locs: Seq[String],
val outputId: Int)
extends Task[U](stageId) {
val split = rdd.splits(partition)
override def run(attemptId: Long): U = {
val context = new TaskContext(stageId, partition, attemptId)
val result = func(context, rdd.iterator(split, context))
context.executeOnCompleteCallbacks()
result
}
override def preferredLocations: Seq[String] = locs
override def toString = "ResultTask(" + stageId + ", " + partition + ")"
}
|