- 概述
说明：代码为tag v3.0.0 preview2时master分支
可以了解：
1.依赖的构建。
2.分区的计算。
3.作业提交模型
4.stage划分细节
5.task分配细节
6.spark处理数据模型
....
最初目的是想了解spark具体是怎样读取数据。但是单一的去看textFile算子代码，发现只是相继new出两个对象hadoopRDD和MapPartitionsRDD，没有调用读数据的具体操作。
然后为了一探究竟，以没有shuffle的简单3行代码为例， 分析其执行以及读数逻辑。
val rddFile = sc.textFile("...")
val rddMap = rdd.map(_.split(","))
print(rddMap.count())
- 代码概述：
textFile会创建HadoopRDD，并初始化它的父类RDD类，然后调用父类的map方法并传入一个只取tuple._2 的函数f，map方法里将用一个新匿名函数来包装 函数f,用来构建MapPartitionsRDD,构建同时也会初始化其父类RDD类，且传入当前rdd用以标识依赖。然后会触发第二行代码map(_.split(","))。此时就会直接调用父类的map方法，传入函数自然就是我们写的split的函数，同样新匿名函数进行包装，构建新的MapPartitionsRDD,初始化父类Rdd，传入依赖rdd。 - 关键代码：
//RDD类
abstract class RDD[T: ClassTag](
@transient private var _sc: SparkContext,
@transient private var deps: Seq[Dependency[_]]
//初始化时传入，还有一辅助构造函数用来将rdd转Seq[Dependency[_]]。
)
//RDD类中的map方法
def map[U: ClassTag](f: T => U): RDD[U] = withScope {
// clean方法实际上调用了ClosureCleaner的clean方法，旨在清除闭包中的不能序列化的变量，防止RDD在网络传输过程中反序列化失败[1]
val cleanF = sc.clean(f)
new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF))
}
//MapPartitionsRDD类
private[spark] class MapPartitionsRDD[U: ClassTag, T: ClassTag](
var prev: RDD[T],
f: (TaskContext, Int, Iterator[T]) => Iterator[U], // (TaskContext, partition index, iterator)
preservesPartitioning: Boolean = false,
isFromBarrier: Boolean = false,
isOrderSensitive: Boolean = false)
extends RDD[U](prev)
说明：第一行代码执行完返回的是MapPartitionsRDD,内部其实还构建了一个HadoopRDD，MapPartitionsRDD的操作其实取value.toString。
-
print(rddMap.count())代码
概述：
1.count内部调用sc.runJob(this, Utils.getIteratorSize _).sum runJob函数会触发DagScheduler去分解任务并提交到TaskScheduler.
2.参数this就是最后调用count操作的rdd,Utils.getIteratorSize一个工具类方法，该方法传入参数为迭代器，方法逻辑 进行迭代器累加，这里其实就是一个partition数据的累加。后面.sum就是每个分区数据sum。
3.注意第二个参数后面有一下划线，是将工具类中定义的方法转为函数，因为函数才能作为参数传递，而方法是不行的。
textFile.png
-
分析runJob：
1.计算分区。
2.提交 job到DAGscheduler. -
1.计算分区,从当前rdd的检查点获取没有则从getPartitions计算，getPartitions会先获取到rdd的依赖，然后再从依赖的rdd中获取partition,也是从检查点或getPartitions计算。递归到第一个HadoopRdd没有依赖时，根据conf，最小分区数，底层数据文件个数以及大小等。获取到InputSplit，代表 hdfs 中的一份数据 ，就是一个 HadoopPartition。其定义了分割的长度及位置。分割长度 是指分割数据的大小（以字节为单位），而分割位置 是分割所在的机器结点名称组成的列表， 分割位置中就能获取到 数据所在的 host 和 rack。
代码:这里获取到分区然后取一个range
def runJob[T, U: ClassTag](rdd: RDD[T], func: Iterator[T] => U): Array[U] = {
runJob(rdd, func, 0 until rdd.partitions.length)
}
rdd_partitions.png
-
2.submit 到 DAGScheduler，先构建JobSubmitted(包括rdd,分区,分区处理函数等信息)，提交到eventProcessLoop内的BlockingQueue。DAGScheduler在初始化时start了一个线程，用以死循环从loop中取事件，进行分事件类型处。并构建一个waiter,监听作业task完成执行resultHandler(将每个分区结果映射到数组)。
postJob.png
代码：
sparkcontext 类
def runJob[T, U: ClassTag](
rdd: RDD[T],//map操作rdd
func: (TaskContext, Iterator[T]) => U,//迭代器累加的函数
partitions: Seq[Int],//partition数量
//将分区结果res，映射到数组index位置匿名函数，
//val results = new Array[U](partitions.size)
//(index, res) => results(index) = res）
resultHandler: (Int, U) => Unit): Unit = {
val callSite = getCallSite//包含最靠近栈顶的用户类及最靠近栈底的Scala或Spark核心类信息
val cleanedFunc = clean(func)
dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
}
//DAGScheduler 类 submitJob在返回waiter
def runJob[T, U](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
callSite: CallSite,
resultHandler: (Int, U) => Unit,
properties: Properties): Unit = {
val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
ThreadUtils.awaitReady(waiter.completionFuture, Duration.Inf)
}
//构建JobSubmitted,并提交到事件循环处理器
def submitJob[T, U](
rdd: RDD[T],//split操作rdd
func: (TaskContext, Iterator[T]) => U,//迭代器累加函数
partitions: Seq[Int],//partition数量
callSite: CallSite,
resultHandler: (Int, U) => Unit,//结果处理句柄函数
properties: Properties): JobWaiter[U] = {
val waiter = new JobWaiter[U](this, jobId, partitions.size, resultHandler)
eventProcessLoop.post(JobSubmitted(
jobId, rdd, func2, partitions.toArray, callSite, waiter,
Utils.cloneProperties(properties)))
waiter
}
- 根据JobSubmitted事件，用handleJobSubmitted处理先划分stage，再按stage顺序依次提交到taskScheduler.
划分stage,
1.createResultStage创建finalStage,在创建finalStage时，调用getorCreate获取父stage,首先依次遍历当前rdd依赖，先找到rdd所有宽依赖,再遍历这些宽依赖。
2.对宽依赖里的rdd继续深度遍历，找到当前rdd所有祖宗的宽依赖。
3.遍历2中所有宽依赖准备创建stage,创建stage时会传入上一个stage。所以会根据当前rdd重新调用getorCreate。递归终止就是父stage为空返回，创建第一个stage.然后第一个stage返回创建第二个...。
4.递归创建完后，会返回一个stage,然后根据最后一个rdd创建finalstage.
说明：每个stage包含其父stage，包含宽依赖信息，分区信息。
createStage.png
代码
//开始创建stage
finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite)
//创建stage前，先获取父stage.
private def createResultStage(...): ResultStage = {
val parents = getOrCreateParentStages(rdd, jobId)
val stage = new ResultStage(id, rdd, func, partitions, parents, jobId, callSite)
stage
}
//1.先获取当前rdd直系父宽依赖，也就是他的爷爷宽依赖是获取不到的getShuffleDependencies。
//2.遍历依赖获取stagegetOrCreateShuffleMapStage
private def getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
getShuffleDependencies(rdd).map { shuffleDep =>
getOrCreateShuffleMapStage(shuffleDep, firstJobId)
}.toList
}
//create依赖or Get。
//1.create时先拿到所有祖宗依赖，指深度遍历，所有宽依赖子节点，都会拿到。并遍历所有宽依赖，创建stage.
//2.根据当前宽依赖的rdd的所有祖宗依赖创建完后，会g根据当前宽依赖创建stage
private def getOrCreateShuffleMapStage(...)ShuffleMapStage={
shuffleIdToMapStage.get(shuffleDep.shuffleId) match {
case Some(stage) =>
stage
case None =>
//拿到所有祖宗依赖，然后遍历
getMissingAncestorShuffleDependencies(shuffleDep.rdd).foreach { dep =>
if (!shuffleIdToMapStage.contains(dep.shuffleId)) {
createShuffleMapStage(dep, firstJobId)
}
}
// Finally, create a stage for the given shuffle
//根据当前宽依赖创建stage
dependency.createShuffleMapStage(shuffleDep, firstJobId)
}
}
//创建stage，
//1.拿到需要宽依赖中的rdd,调用getOrCreateParentStages获取父stage。然后创建ShuffleMapStage。
def createShuffleMapStage(...): ShuffleMapStage = {
val rdd = shuffleDep.rdd
//先获取父stage,获取不到时，为空(第一个stage)
val parents = getOrCreateParentStages(rdd, jobId)
val stage = new ShuffleMapStage(
id, rdd, numTasks, parents, jobId, rdd.creationSite, shuffleDep, mapOutputTracker)
stage
}
- 在handleJobSubmitted中创建完stage，然后更新各辅助变量，最后submitStage。提交过程也是递归提交，先递归到底找到没有父宽依赖的stage,进行提交。
private def submitStage(stage: Stage): Unit = {
//更具最后一个stage找打所有stage,并按id排序
val missing = getMissingParentStages(stage).sortBy(_.id)
//如果没有父stage,真正提交到taskScheduler.
if (missing.isEmpty) {
submitMissingTasks(stage, jobId.get)
} else {
//找到的父stage,再次递归调用submitStage,进行查找再提交
for (parent <- missing) {
submitStage(parent)
}
}
}
- submitMissingTasks 提交stage,会在DAGscheduler中将stage转为task,提交到TaskScheduler处理。
1.计算task偏好位置，为stage内的每个分区计算.计算好的位置，将在后面用来分配task在哪执行。
偏好位置为TaskLocation，一个trait，有三个实现类，代表数据存储在不同的位置：
数据存储在Executor内存中，即Partition被cache到了内存中(返回executorId+host)
数据存储在HDFS上(返回host)
数据存储在host这个节点的磁盘上(返回“hdfs_cache”+host)
rdd顺着窄依赖， 往上找父依赖， 直到找到第一个窄依赖， 也就是找数据读取源头。这里是HadoopRDD， 那么每个 task 处理的数据就是一个 HadoopPartition， 其代表 hdfs 中的一份数据 InputSplit， 定义了分割的长度及位置。
读取的是 shuffle 的数据，那么会根据其shufflePartition去查找上个stage写出数据的位置。map写阶段根据分区器生成多少个分区。shufflePartition根据这多少个分区生成一个分区数组。
2.先序列化stage和依赖信息，再构建task,最后用taskSet封装其stage所有task提交到taskScheduler
//计算每个stage里每个分区的位置偏好
val taskIdToLocations: Map[Int, Seq[TaskLocation]] = try {
stage match {
case s: ShuffleMapStage =>
partitionsToCompute.map { id => (id, getPreferredLocs(stage.rdd, id))}.toMap
case s: ResultStage =>
partitionsToCompute.map { id =>
val p = s.partitions(id)
(id, getPreferredLocs(stage.rdd, p))
}.toMap
}
//序列化stage中最后一个rdd,和依赖信息
//构建task，为每个分区分别构建不同类型的task
val tasks: Seq[Task[_]] = try {
stage match {
case stage: ShuffleMapStage =>
partitionsToCompute.map { id =>
new ShuffleMapTask(stage.id,
taskBinary, part, locs, properties,...)
}
case stage: ResultStage =>
partitionsToCompute.map { id =>
new ResultTask(stage.id,
taskBinary, part, locs, ....)
}
}
}
//最后用taskSet封装提交到taskScheduler
taskScheduler.submitTasks(new TaskSet(
tasks.toArray, stage.id, stage.latestInfo.attemptNumber, jobId, properties))
-
此时，已经到了taskScheduler.
1.为当前TaskSet创建TaskSetManager，TaskSetManager负责管理一个taskSet的每一个task，管理task分配，重试，延迟调度等。(创建)
2.当前taskSet添加到调度池中 ,调度池有两个实现FIFOSchedulerBuilder,FairSchedulerBuilder，并且默认是FIFO。
taskManager.png
关键代码：
//TaskSchedulerImpl类
override def submitTasks(taskSet: TaskSet): Unit = {
val tasks = taskSet.tasks
//创建TaskSet的Manager
val manager = createTaskSetManager(taskSet, maxTaskFailures)
//将Manager加入调度池。
schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
//调用CoarseGrainedSchedulerBackend的reviveOffers方法向driveEndPoint发送消息获取资源并计算task执行位置，最后LaunchTask
backend.reviveOffers()
}
//TaskManager类
//创建时会将taskSet里的task按偏好位置加入各pendingTask
addPendingTasks()
private def addPendingTasks(): Unit = {
for (i <- (0 until numTasks).reverse) {
addPendingTask(i, resolveRacks = false)
}
}
private[spark] def addPendingTask(
index: Int,
resolveRacks: Boolean = true,
speculatable: Boolean = false): Unit = {
val pendingTaskSetToAddTo = pendingTask(包含各本地性级别的pendingtask)
for (loc <- tasks(index).preferredLocations) {
loc match {
case e: ExecutorCacheTaskLocation =>
//task的偏好位置有execId，将task的index加入forExecutor的pendingtask
pendingTaskSetToAddTo.forExecutor.getOrElseUpdate(e.executorId, new ArrayBuffer) += index
case e: HDFSCacheTaskLocation =>
//HDFS里也会判断task位置偏好的host是否更封装的资源有一样的host，有则拿出host主机里的executor，去查看task里是否有对应execId,有则加入forExecutor的pendingtask
val exe = sched.getExecutorsAliveOnHost(loc.host)
exe match {
case Some(set) =>
for (e <- set) {
pendingTaskSetToAddTo.forExecutor.getOrElseUpdate(e, new ArrayBuffer) += index
case _ =>
}
//遍历完task的偏好位置，会将所有task加入forHost的pendingtask。表示每个task都会有host本地性级别
pendingTaskSetToAddTo.forHost.getOrElseUpdate(loc.host, new ArrayBuffer) += index
//解析机架默认为false,将机架加入pendingtask
if (resolveRacks) {
sched.getRackForHost(loc.host).foreach { rack =>
pendingTaskSetToAddTo.forRack.getOrElseUpdate(rack, new ArrayBuffer) += index
}
}
}
//task偏好位置为Nil在会加入noPrefs的pendingtask
if (tasks(index).preferredLocations == Nil) {
pendingTaskSetToAddTo.noPrefs += index
}
//会将所有task加入all的pendingtask。
pendingTaskSetToAddTo.all += index
}
3.加入调度池后 backend.reviveOffers()调用了CoarseGrainedSchedulerBackend的reviveOffers方法向driveEndPoint发送消息获取资源(基于事件模型的调用，reviveOffers事件有两种触发模式：1.周期性触发的，默认1秒一次。2.被TaskSchedulerImpl里用backend.reviveOffers()调用)。触发后调用makeOffers(),a.先过滤出活跃的executor并封装成WorkerOffer(cores,host,execId,..)。b.然后根据resourceOffers按资源和task本地性找出最佳执行策略，返回Seq[TaskDescription]task的描述信息。最后交给SchedulerBackend发送task的描述信息到描述里的executor上执行
//CoarseGrainedSchedulerBackend类
private def makeOffers(): Unit = {
val taskDescs = withLock {
//过滤资源
val activeExecutors = executorDataMap.filterKeys(executorIsAlive)
//整理成workOffers
val workOffers = activeExecutors.map {
case (id, executorData) =>
new WorkerOffer(id, executorData.executorHost, executorData.freeCores,
Some(executorData.executorAddress.hostPort),
executorData.resourcesInfo.map { case (rName, rInfo) =>
(rName, rInfo.availableAddrs.toBuffer)
})
}.toIndexedSeq
//找出要在哪些Worker上启动哪些task
scheduler.resourceOffers(workOffers)
}
//对返回的taskDesc发送到对应Executor执行task
if (taskDescs.nonEmpty) {
launchTasks(taskDescs)
}
}
//TaskScheduler类，查找task最佳资源
//代码非常长，套用方法非常多，只显示核心逻辑，避免文章过长
//提示:伪代码
遍历排序好的TaskSet,这里其实就是taskManager.
for (taskSet <- sortedTaskSets) {
再遍历taskSet里拥有的级别，从最优级别开始
for (currentMaxLocality <- taskSet.myLocalityLevels) {
var launchedTask =false
do {
//找不到task执行资源为false
//查找资源
launchedTask = resourceOfferSingleTaskSet(taskSet,
currentMaxLocality, shuffledOffers,tasks,...)
} while (launchedTask)
}
private def resourceOfferSingleTaskSet(....){
//遍历每个workOffer(资源)
for (i <- 0 until shuffledOffers.size) {
//空闲资源大于task执行需要的资源
if (availableCpus(i) >= CPUS_PER_TASK){
//resourceOffer方法为去查找最佳task执行位置，返回类型Option[taskDesc]
for (task <- taskSet.resourceOffer(execId, host, maxLocality)) {
//各种信息更新
}
}
}
return launchedTask
}
def resourceOffer(
execId: String,
host: String,
maxLocality: TaskLocality.TaskLocality,
availableResources: Map[String, Seq[String]] = Map.empty)
: Option[TaskDescription] =
{
//遍历的最优数据本地性不为NO_PREF,计算一个允许的最低本地性级别
if (maxLocality != TaskLocality.NO_PREF) {
allowedLocality = getAllowedLocalityLevel(curTime)
if (allowedLocality > maxLocality) {
// We're not allowed to search for farther-away tasks
allowedLocality = maxLocality
}
}
//dequeueTask查找task的index,返回类型Option[int]
dequeueTask(execId, host, allowedLocality).map { case ((index, taskLocality, speculative)) =>
//找到的task，进行封装taskDesc，将资源的地址等等信息封装。并返回
return TaskDesc
}
}
private def dequeueTask(
execId: String,
host: String,
maxLocality: TaskLocality.Value): Option[(Int, TaskLocality.Value, Boolean)] = {
//pendingTask
val pendingTaskSetToUse = if (speculative) pendingSpeculatableTasks else pendingTasks
//dequeue方法主要是dequeueTaskFromList从 pendingTask取出task的Index，返回类型option[int]
def dequeue(list: ArrayBuffer[Int]): Option[Int] = {
val task = dequeueTaskFromList(execId, host, list, speculative)
if (speculative && task.isDefined) {
speculatableTasks -= task.get
}
task
}
//最先默认从forExecutor根据资源的execId查找task
dequeue(pendingTaskSetToUse.forExecutor.getOrElse(execId, ArrayBuffer())).foreach { index =>
return Some((index, TaskLocality.PROCESS_LOCAL, speculative))
}
//比较允许的最低级别大于Node_local级别,通过主机名找到相应的Task
if (TaskLocality.isAllowed(maxLocality, TaskLocality.NODE_LOCAL)) {
dequeue(pendingTaskSetToUse.forHost.getOrElse(host, ArrayBuffer())).foreach { index =>
return Some((index, TaskLocality.NODE_LOCAL, speculative))
}
}
//node_local之后，会比较允许的最低级别大于NO_PREF级别,通过noPrefs去pendingTask查找
if (TaskLocality.isAllowed(maxLocality, TaskLocality.NO_PREF)) {
dequeue(pendingTaskSetToUse.noPrefs).foreach { index =>
return Some((index, TaskLocality.PROCESS_LOCAL, speculative))
}
}
//NO_PREF之后，会比较允许的最低级别大于RACK_LOCAL级别,通过rack去forRack pendingTask查找
if (TaskLocality.isAllowed(maxLocality, TaskLocality.RACK_LOCAL)) {
for {
rack <- sched.getRackForHost(host)
index <- dequeue(pendingTaskSetToUse.forRack.getOrElse(rack, ArrayBuffer()))
} {
return Some((index, TaskLocality.RACK_LOCAL, speculative))
}
}
//最后 去ANY pendingTask查找
if (TaskLocality.isAllowed(maxLocality, TaskLocality.ANY)) {
dequeue(pendingTaskSetToUse.all).foreach { index =>
return Some((index, TaskLocality.ANY, speculative))
}
}
None
}
task_arrange.png
- launchTasks
- makeOffers方法将task分配资源后，调用launchTasks，发送到指定的executor执行task,先对taskDesc消息序列化,可以在网络间进行传输。再获取exector信息，然后发送LaunchTask消息执行task
private def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
for (task <- tasks.flatten) {
//首先对每个executor需要执行的task消息序列化一下，可以在网络间进行传输
val serializedTask = TaskDescription.encode(task)
if (serializedTask.limit >= maxRpcMessageSize) {
//根据task消息中的executorId找到运行的executor
val executorData = executorDataMap(task.executorId)
//并将executor空余的core数减去自身需要的core数
executorData.freeCores -= scheduler.CPUS_PER_TASK
//向executor发送LaunchTask消息，用于在对应executor上启动task
executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))
}
}
}
- Executor端接收LaunchTask事件
driver端向executor发送任务需要通过后台辅助进程CoarseGrainedSchedulerBackend，那么自然而然executor接收任务也有对应的后台辅助进程CoarseGrainedExecutorBackend，该进程与executor一一对应，提供了executor和driver通讯的功能。下面看CoarseGrainedExecutorBackend接收到事件后是如何处理的：
1.将TaskDescription反序列化
2.调用executor的launchTask执行task
case LaunchTask(data) =>
if (executor == null) {
exitExecutor(1, "Received LaunchTask command but executor was null")
} else {
// 将TaskDescription反序列化
val taskDesc = ser.deserialize[TaskDescription](data.value)
// 调用executor的launchTask来加载该task
executor.launchTask(this, taskId = taskDesc.taskId, attemptNumber = taskDesc.attemptNumber,
taskDesc.name, taskDesc.serializedTask)
}
- executor接受task，创建了一个TaskRunner（继承于 Runnable）并加入到线程池中执行。其run方法:反序列化得到task各信息，然后调用task的run方法，根据task的类型(shuffle,result)真正执行task，执行完后。清除分配内存，然后序列化task的结果，包装成directResult，再次序列化，根据其结果大小将结果以不同的方式返回给driver
def launchTask(
context: ExecutorBackend,
taskId: Long,
attemptNumber: Int,
taskName: String,
serializedTask: ByteBuffer): Unit = {
// 创建一个TaskRunner
val tr = new TaskRunner(context, taskId = taskId, attemptNumber = attemptNumber, taskName,
serializedTask)
runningTasks.put(taskId, tr)
// 将tr放到线程池中执行
threadPool.execute(tr)
}
- ShuffleMapTask：
shuffleMapTask的输出直接通过Shuffle write写磁盘，为下游的stage的Shuffle Read准备数据
反序列化出rdd和ShuffleDependency，获取ShuffleManager的writer将一个rdd的某个分区写入到磁盘，通过rdd的iterator方法能遍历对应分区的数据。
override def runTask(context: TaskContext): MapStatus = {
// 反序列化出rdd和ShuffleDependency
val (rdd, dep) = ser.deserialize[(RDD[_], ShuffleDependency[_, _, _])](
ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)
_executorDeserializeTime = System.currentTimeMillis() - deserializeStartTime
var writer: ShuffleWriter[Any, Any] = null
try {
// 获取shuffleManager
val manager = SparkEnv.get.shuffleManager
// 通过shuffleManager的getWriter()方法，获得shuffle的writer
writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId, context)
// 通过rdd指定分区的迭代器iterator方法来遍历每一条数据，再之上再调用writer方法以写数据
writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])
}
}
- ResultTask：
1.反序列化得到rdd和func(count操作传进来的迭代器累加匿名函数)，执行func，传入rdd调用iterator方法获取到数据的迭代器。func对迭代器累计。
override def runTask(context: TaskContext): U = {
val deserializeStartTime = System.currentTimeMillis()
val ser = SparkEnv.get.closureSerializer.newInstance()
// 反序列化
val (rdd, func) = ser.deserialize[(RDD[T], (TaskContext, Iterator[T]) => U)](
ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)
_executorDeserializeTime = System.currentTimeMillis() - deserializeStartTime
// 对rdd的指定分区的迭代器执行func函数，并返回结果
func(context, rdd.iterator(partition, context))
}
- shuffleTask 后序再将shuffle的时候再补充，这里主要讲ResultMaptask。关键一行代码 func(context, rdd.iterator(partition, context))
rdd调用iterator获取该分区的迭代器，以用来执行最后的func。
如何获取的迭代器：
判断rdd是否缓存，checkPoint，没有则依次从最后一个rdd向上调用compute获取依赖。在我们的测试例子中，最后一个是split操作的MapPartitionRDD,所以先调用此rdd的compute
//MapPartitionRDD
override def compute(split: Partition, context: TaskContext): Iterator[U] =
f(context, split.index, firstParent[T].iterator(split, context))
看代码：compute函数是调用f函数获取迭代器，f函数是一个匿名函数。在这就是split操作。也就是说获取的迭代器在此做一个split操作，在返回。
然后继续看 firstParent[T].iterator。调用上一个rdd的iterator方法获取迭代器，其实就是跟刚刚一样了。直到获取到第一个HadoopRDD调用compute方法计算当前partition的Iterator.。
- HadoopRDD compute
创建一个迭代器，内部用inputFormat和分区构造一个reader,利用reader重写迭代器next方法用以读取数据。hadoopRDD默认的inputFormat是FileinputFormat将数据读成分割成一行.reader是LineRecordReader,key为偏移量，一行为value。
最后返回数据的迭代器。
override def compute(theSplit: Partition, context: TaskContext): InterruptibleIterator[(K, V)] = {
val iter = new NextIterator[(K, V)] {
//将compute的输入theSplit，转换为HadoopPartition
private val split = theSplit.asInstanceOf[HadoopPartition]
logInfo("Input split: " + split.inputSplit)
private val jobConf = getJobConf()
...
//创建reader
private var reader: RecordReader[K, V] = null
//先根据conf拿到InputFormat，
private val inputFormat = getInputFormat(jobConf)
//从InputFormat中getRecordReader,传入HadoopPartition，conf。
reader =
try {
inputFormat.getRecordReader(split.inputSplit.value, jobConf, Reporter.NULL)
} catch {... }
...
private val key: K = if (reader == null) null.asInstanceOf[K] else reader.createKey()
private val value: V = if (reader == null) null.asInstanceOf[V] else reader.createValue()
//重写next方法，用以遍历数据
override def getNext(): (K, V) = {
try {
finished = !reader.next(key, value)
} catch {... }
if (!finished) {
inputMetrics.incRecordsRead(1)
}
(key, value)
}
//最后构建一个包装好的迭代器，传入根据reader读数，并重写了next方法的迭代器。
new InterruptibleIterator[(K, V)](context, iter)
}
- 迭代器，返回至task执行时func(context, rdd.iterator(partition, context))，执行func函数，对迭代器里的数据遍历累加，再从第一个rdd的迭代器遍历出数据时，也会作用在后面rdd的f函数上，也就是用户编写的操作。
func 代码：在count提交作业时，传入
//count提交作业，传入Utils.getIteratorSize _ 对没每个分区的计算。
def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum
//func函数 就是累加迭代器，触发整个数据操作。
def getIteratorSize(iterator: Iterator[_]): Long = {
var count = 0L
while (iterator.hasNext) {
count += 1L
iterator.next()
}
count
}
总流程：概括
all_operator.png
网友评论