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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
|
package spark.scheduler
import spark.serializer.SerializerInstance
import java.io.{DataInputStream, DataOutputStream}
import java.nio.ByteBuffer
import it.unimi.dsi.fastutil.io.FastByteArrayOutputStream
import spark.util.ByteBufferInputStream
import scala.collection.mutable.HashMap
import spark.executor.TaskMetrics
/**
* A task to execute on a worker node.
*/
private[spark] abstract class Task[T](val stageId: Int) extends Serializable {
def run(attemptId: Long): T
def preferredLocations: Seq[String] = Nil
var generation: Long = -1 // Map output tracker generation. Will be set by TaskScheduler.
var metrics: Option[TaskMetrics] = None
}
/**
* Handles transmission of tasks and their dependencies, because this can be slightly tricky. We
* need to send the list of JARs and files added to the SparkContext with each task to ensure that
* worker nodes find out about it, but we can't make it part of the Task because the user's code in
* the task might depend on one of the JARs. Thus we serialize each task as multiple objects, by
* first writing out its dependencies.
*/
private[spark] object Task {
/**
* Serialize a task and the current app dependencies (files and JARs added to the SparkContext)
*/
def serializeWithDependencies(
task: Task[_],
currentFiles: HashMap[String, Long],
currentJars: HashMap[String, Long],
serializer: SerializerInstance)
: ByteBuffer = {
val out = new FastByteArrayOutputStream(4096)
val dataOut = new DataOutputStream(out)
// Write currentFiles
dataOut.writeInt(currentFiles.size)
for ((name, timestamp) <- currentFiles) {
dataOut.writeUTF(name)
dataOut.writeLong(timestamp)
}
// Write currentJars
dataOut.writeInt(currentJars.size)
for ((name, timestamp) <- currentJars) {
dataOut.writeUTF(name)
dataOut.writeLong(timestamp)
}
// Write the task itself and finish
dataOut.flush()
val taskBytes = serializer.serialize(task).array()
out.write(taskBytes)
out.trim()
ByteBuffer.wrap(out.array)
}
/**
* Deserialize the list of dependencies in a task serialized with serializeWithDependencies,
* and return the task itself as a serialized ByteBuffer. The caller can then update its
* ClassLoaders and deserialize the task.
*
* @return (taskFiles, taskJars, taskBytes)
*/
def deserializeWithDependencies(serializedTask: ByteBuffer)
: (HashMap[String, Long], HashMap[String, Long], ByteBuffer) = {
val in = new ByteBufferInputStream(serializedTask)
val dataIn = new DataInputStream(in)
// Read task's files
val taskFiles = new HashMap[String, Long]()
val numFiles = dataIn.readInt()
for (i <- 0 until numFiles) {
taskFiles(dataIn.readUTF()) = dataIn.readLong()
}
// Read task's JARs
val taskJars = new HashMap[String, Long]()
val numJars = dataIn.readInt()
for (i <- 0 until numJars) {
taskJars(dataIn.readUTF()) = dataIn.readLong()
}
// Create a sub-buffer for the rest of the data, which is the serialized Task object
val subBuffer = serializedTask.slice() // ByteBufferInputStream will have read just up to task
(taskFiles, taskJars, subBuffer)
}
}
|
