Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/1498#discussion_r15567033
--- Diff: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
---
@@ -691,47 +689,81 @@ class DAGScheduler(
}
}
-
/** Called when stage's parents are available and we can now do its
task. */
private def submitMissingTasks(stage: Stage, jobId: Int) {
logDebug("submitMissingTasks(" + stage + ")")
// Get our pending tasks and remember them in our pendingTasks entry
stage.pendingTasks.clear()
var tasks = ArrayBuffer[Task[_]]()
+
+ val properties = if (jobIdToActiveJob.contains(jobId)) {
+ jobIdToActiveJob(stage.jobId).properties
+ } else {
+ // this stage will be assigned to "default" pool
+ null
+ }
+
+ runningStages += stage
+ // SparkListenerStageSubmitted should be posted before testing whether
tasks are
+ // serializable. If tasks are not serializable, a
SparkListenerStageCompleted event
+ // will be posted, which should always come after a corresponding
SparkListenerStageSubmitted
+ // event.
+ listenerBus.post(SparkListenerStageSubmitted(stage.info, properties))
+
+ // TODO: Maybe we can keep the taskBinary in Stage to avoid
serializing it multiple times.
+ // Broadcasted binary for the task, used to dispatch tasks to
executors. Note that we broadcast
+ // the serialized copy of the RDD and for each task we will
deserialize it, which means each
+ // task gets a different copy of the RDD. This provides stronger
isolation between tasks that
+ // might modify state of objects referenced in their closures. This is
necessary in Hadoop
+ // where the JobConf/Configuration object is not thread-safe.
+ var taskBinary: Broadcast[Array[Byte]] = null
+ try {
+ // For ShuffleMapTask, serialize and broadcast (rdd, shuffleDep).
+ // For ResultTask, serialize and broadcast (rdd, func).
+ val taskBinaryBytes: Array[Byte] =
+ if (stage.isShuffleMap) {
+ Utils.serializeTaskClosure((stage.rdd, stage.shuffleDep.get) :
AnyRef)
+ } else {
+ Utils.serializeTaskClosure((stage.rdd,
stage.resultOfJob.get.func) : AnyRef)
+ }
+ taskBinary = sc.broadcast(taskBinaryBytes)
+ } catch {
+ // In the case of a failure during serialization, abort the stage.
+ case e: NotSerializableException =>
+ abortStage(stage, "Task not serializable: " + e.toString)
+ runningStages -= stage
+ return
+ case NonFatal(e) =>
+ abortStage(stage, s"Task serialization failed:
$e\n${e.getStackTraceString}")
+ runningStages -= stage
+ return
+ }
+
if (stage.isShuffleMap) {
for (p <- 0 until stage.numPartitions if stage.outputLocs(p) == Nil)
{
val locs = getPreferredLocs(stage.rdd, p)
- tasks += new ShuffleMapTask(stage.id, stage.rdd,
stage.shuffleDep.get, p, locs)
+ val part = stage.rdd.partitions(p)
+ tasks += new ShuffleMapTask(stage.id, taskBinary, part, locs)
}
} else {
// This is a final stage; figure out its job's missing partitions
val job = stage.resultOfJob.get
for (id <- 0 until job.numPartitions if !job.finished(id)) {
- val partition = job.partitions(id)
- val locs = getPreferredLocs(stage.rdd, partition)
- tasks += new ResultTask(stage.id, stage.rdd, job.func, partition,
locs, id)
+ val p: Int = job.partitions(id)
+ val part = stage.rdd.partitions(p)
+ val locs = getPreferredLocs(stage.rdd, p)
+ tasks += new ResultTask(stage.id, taskBinary, part, locs, id)
}
}
- val properties = if (jobIdToActiveJob.contains(jobId)) {
- jobIdToActiveJob(stage.jobId).properties
- } else {
- // this stage will be assigned to "default" pool
- null
- }
-
if (tasks.size > 0) {
- runningStages += stage
- // SparkListenerStageSubmitted should be posted before testing
whether tasks are
- // serializable. If tasks are not serializable, a
SparkListenerStageCompleted event
- // will be posted, which should always come after a corresponding
SparkListenerStageSubmitted
- // event.
- listenerBus.post(SparkListenerStageSubmitted(stage.info, properties))
-
// Preemptively serialize a task to make sure it can be serialized.
We are catching this
// exception here because it would be fairly hard to catch the
non-serializable exception
// down the road, where we have several different implementations
for local scheduler and
// cluster schedulers.
+ //
+ // We've already serialized RDDs and closures in taskBinary, but
here we check for all other
+ // objects such as Partition.
try {
SparkEnv.get.closureSerializer.newInstance().serialize(tasks.head)
--- End diff --
That one also does compression. But honestly given they are both one-liner,
I'm not sure if it is worth it ...
---
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.
---
