热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

[第九章]Job触发流程原理剖析

上一节我们是不是讲到,Driver,Application注册到Master上面后,Master中调用scheduler()进行资源调度,在这个里面通过LaunchDriver()

上一节我们是不是讲到,Driver,Application注册到Master上面后,Master中调用scheduler()进行资源调度,在这个里面通过LaunchDriver(),LaunchExecutor(),向Worker发出启动Driver,Exeutor的请求(或者说命令),

Worker接收到发来的请求,通过创建DriverRunner,ExecutorRunner线程来启动我们的Driver与Application.在启动完成后,根据第一章节我们分析的Spark核心原理,Executor会反向注册到Driver上,这样Driver就清楚哪些Executor在执行Application,实际上到了这个时候,我们的SparkContext已经全部初使化完成。

接下来我们就要继续执行我们自己编写的代码程序,其实一个Application包括多个JOB,那么JOB是如何划分的呢? 实际上一个Action操作,会划分一个JOB,就是说多个Action操作就会有多个JOB,JOB执行的顺序是从第一个开始。
下面我们以wordcount实例来详细分析一下JOB的划分:

val lines = sc.textFile(...)
val words = lines.flatMap(line => line.split(" "))
val pairs = words.map(word => (word, 1))
val counts = pairs.reduceByKey(_ + _)
counts.foreach(count => println(count._1 + ": " + count._2))

看到上面的几行代码,大家是不是太熟悉了,这就是我们学习spark的第一个例子。
首先我们先来分析第一行代码:
val lines = sc.textFile()

def textFile(path: String, minPartitions: Int = defaultMinPartitions): RDD[String] = {
assertNotStopped()
hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
minPartitions).map(pair => pair._2.toString).setName(path)
}

上面的代码是不是很熟悉,

  • 首先,textFile会调用一个hadoopFile(…)的方法,我们看里面的参数,TextInputFormat这是不是很熟悉,这是hadoop里读取文本文件的解析器啊
  • 后面的classOf[LongWritable], classOf[Text]这就是hadoop
    map()函数的k1,v1吧
    这可都是hadoop里的东东。下面我们看hadoopFile方法的代码

def hadoopFile[K, V](
path: String,
inputFormatClass: Class[_ <: InputFormat[K, V]],
keyClass: Class[K],
valueClass: Class[V],
minPartitions: Int = defaultMinPartitions
): RDD[(K, V)] = {
assertNotStopped()
// A Hadoop configuration can be about 10 KB, which is pretty big, so broadcast it.
//这个是不是我们把hadoop配置数据序列化后在广播出去,共享广播这个我们在之前讲过,不会忘记吧!
val cOnfBroadcast= broadcast(new SerializableWritable(hadoopConfiguration))
val setInputPathsFunc = (jobConf: JobConf) => FileInputFormat.setInputPaths(jobConf, path)
new HadoopRDD(
this,
confBroadcast,
Some(setInputPathsFunc),
inputFormatClass,
keyClass,
valueClass,
minPartitions).setName(path)
}

不难看出,这个方法返回了一个HadoopRDD算子
接着我们看后面调用了map(&#8230;),map方法具体我们这里就不分析,以前的章节我们分析过了,
.map(pair => pair._2.toString).setName(path)
其中的pair是不是一个tuple,也就是我们Hadoop的k1,v1,那么pair._2.toStrig,是不是我们读取文件的每一行内容。

接下来我们继续执行代码:

val words = lines.flatMap(line => line.split(" "))
val pairs = words.map(word => (word, 1))

这两行与上面分析map的相似,我们就不重复了。接着我们执行

val counts = pairs.reduceByKey(_ + _)

这行代码我们惊奇的发现在RDD类里没有reduceByKey方法,这是为什么了?没有方法如何调用。
大家是不是想到scala中的隐式转换,在RDD类中找查到了隐式转换方法rddToPairRDDFunctions(),把rdd转换成了rddToPairRDDFunctions,返回这个对象,如下面代码

implicit def rddToPairRDDFunctions[K, V](rdd: RDD[(K, V)])
(implicit kt: ClassTag[K], vt: ClassTag[V], ord: Ordering[K] = null): PairRDDFunctions[K, V] = {
new PairRDDFunctions(rdd)
}

果然在PairRDDFunction类中找到这个方法:

def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)] = {
combineByKey[V]((v: V) => v, func, func, partitioner)
}`

接下来我们执行最后一行代码,

counts.foreach(count => println(count._1 + ": " + count._2))

当我们看foreach时,明白这就是一个action算子。它调用了sparkContext的runJob方法,来划分一个JOB

def foreach(f: T => Unit) {
val cleanF = sc.clean(f)
//执行SparkContext中的runJob
sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))
}

经过多个runJob()重载的调用,最后我们找到最终的调用方法:

def runJob[T, U: ClassTag](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
allowLocal: Boolean,
resultHandler: (Int, U) => Unit) {
if (stopped) {
throw new IllegalStateException("SparkContext has been shutdown")
}
val callSite = getCallSite
val cleanedFunc = clean(func)
logInfo("Starting job: " + callSite.shortForm)
if (conf.getBoolean("spark.logLineage", false)) {
logInfo("RDD's recursive dependencies:\n" + rdd.toDebugString)
}
//调用sparkContext之前初使化 时创建的DAGScheduler的runJob方法
dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, allowLocal,
resultHandler, localProperties.get)
progressBar.foreach(_.finishAll())
rdd.doCheckpoint()

看到这里大家 是不是明白了,最终其实就是调用了sparkContext之前初使化 时创建的DAGScheduler的runJob方法,里面的第一个参数rdd是不是以前的rdd算子。
这里也说明了JOB的划分都是在DAGScheduler里完成的。接下来我们会在在下一章节会对DAGScheduler如何划分JOB进行深入的分析.

本章中每一个字(包括源码注解)都是作者敲出来的,你感觉有用,帮点击&#8217;喜欢&#8217;


推荐阅读
author-avatar
1911530988com
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有