1、概述 java
以前几篇文章对Spark集群的Master、Worker启动流程进行了源码剖析，后面直接从客户端角度出发，讲解了spark-submit任务提交过程及driver的启动；集群启动、任务提交、SparkContext初始化等前期准备工做完成以后，后面就是咱们的主函数的代码Job如何触发的，本篇文章仍是结合源码进行剖析。
sql
软件版本：
apache
spark2.2.0
api
2、Job触发流程源码剖析数组
1. 咱们先上一段最简单的代码，读取本地文件进行WordCount，并打印统计结果，代码以下：
缓存
package com.hadoop.ljs.spark220.study;import org.apache.spark.SparkConf;import org.apache.spark.api.java.JavaPairRDD;import org.apache.spark.api.java.JavaRDD;import org.apache.spark.api.java.JavaSparkContext;import org.apache.spark.api.java.function.FlatMapFunction;import org.apache.spark.api.java.function.Function2;import org.apache.spark.api.java.function.PairFunction;import org.apache.spark.api.java.function.VoidFunction;import org.apache.spark.sql.SparkSession;import scala.Tuple2;import java.util.Arrays;import java.util.Iterator;/** * @author: Created By lujisen * @company ChinaUnicom Software JiNan * @date: 2020-03-12 08:26 * @version: v1.0 * @description: com.hadoop.ljs.spark220.study */public class Example1 { public static void main(String[] args) throws Exception{ /*spark环境初始化*/ SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("Example1"); SparkSession sc = SparkSession.builder().config(sparkConf).getOrCreate(); JavaSparkContext jsc = new JavaSparkContext(sc.sparkContext()); /*读取本地文件*/ JavaRDD<String> sourceRDD = jsc.textFile("D:\\kafkaSSL\\kafka_client_jaas.conf"); /*转换多维为一维数组*/ JavaRDD<String> words = sourceRDD.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) { return Arrays.asList(s.split(" ")).iterator(); } }); /*转换成（hello,1)格式*/ JavaPairRDD<String, Integer> wordOne = words.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) { return new Tuple2<String, Integer>(s, 1); } }); /*根据key进行聚合*/ JavaPairRDD<String, Integer> wordCount = wordOne.reduceByKey(new Function2<Integer, Integer, Integer>() { @Override public Integer call(Integer v1, Integer v2) { return v1+v2; } }); /*打印结果*/ wordCount.foreach(new VoidFunction<Tuple2<String, Integer>>() { @Override public void call(Tuple2<String, Integer> result){ System.out.println("word: "+result._1+" count: "+result._2); } });
}}
咱们一行行的进行分析，首先看读取本地文件textFile()函数：
ide
/*这里直接调用的SparkContext的textFile函数*/ def textFile(path: String): JavaRDD[String] = sc.textFile(path)
2. 直接看sc.textFile()函数:函数
def textFile( path: String, minPartitions: Int = defaultMinPartitions): RDD[String] = withScope { assertNotStopped() /*这里调用了hadoopFile函数,传入三个，写过Mapreuce的时候都知道 第二个参数就是Map的输入格式化类型，参数3是行号 4是一行的内容*/ /*hadoopFile()函数，返回了一个HadoopRDD*/ hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text], minPartitions).map(pair => pair._2.toString).setName(path) }
看hadoopFile()函数
oop
def hadoopFile[K, V]( path: String, inputFormatClass: Class[_ <: InputFormat[K, V]], keyClass: Class[K], valueClass: Class[V], minPartitions: Int = defaultMinPartitions): RDD[(K, V)] = withScope { assertNotStopped() // This is a hack to enforce loading hdfs-site.xml. // See SPARK-11227 for details. FileSystem.getLocal(hadoopConfiguration) //这里把hadoopConfiguration配置作了一个广播变量 val confBroadcast = broadcast(new SerializableConfiguration(hadoopConfiguration)) /* 传入一个jobConf对输入数据进行格式化*/ val setInputPathsFunc = (jobConf: JobConf) => FileInputFormat.setInputPaths(jobConf, path) /* 返回一个HadoopRDD实例，这里Hadoop配置文件是以广播变量的方式传进去的*/ /*广播变量 每一个Worker保存一份，被多个Executor共享*/ /*HadoopRDD继承自RDD*/ new HadoopRDD( this, confBroadcast, Some(setInputPathsFunc), inputFormatClass, keyClass, valueClass, minPartitions).setName(path) }
上面直接对HadopRDD作了一个map转换，这里Hadoop继承自RDD,调用的是RDD里面的map()函数，咱们直接看看map函数代码：
post
/* 最后实际上是返回了一个MapPartitionsRDD，里面是（key，value），key是行号，value是内容*/ def map[U: ClassTag](f: T => U): RDD[U] = withScope { val cleanF = sc.clean(f) new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF)) }
上面对返回的RDD是一个<key,value>键值对，而后.map(pair => pair._2.toString对其进行了转换，其实就是去掉了那个key行号，剩下的是一个vlaue数组，里面是每行的内容，至此textFile这一行剖析完毕。
3.主函数的第30-42行都是对RDD进行了一系列的转换，其实都是调用RDD.scala中的内容对MapPartitionsRDD进行的转换，有兴趣你能够跟进去看一下，比较简单：
def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U] = withScope { val cleanF = sc.clean(f) new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.flatMap(cleanF)) }
/* mapToPair函数里面实际上是调用的rdd.map函数，刚才上面已经说过了*/ def mapToPair[K2, V2](f: PairFunction[T, K2, V2]): JavaPairRDD[K2, V2] = { def cm: ClassTag[(K2, V2)] = implicitly[ClassTag[(K2, V2)]] new JavaPairRDD(rdd.map[(K2, V2)](f)(cm))(fakeClassTag[K2], fakeClassTag[V2]) }
4.最后调用reduceBykey进行了聚合，这里就比较重要了，咱们以前讲过一个spark任务里面会有多个job,job的划分依据是action，有几个action就有几个job,而每一个job的划分依据是shuffle，只要发生了shuffle就会有新的stage生成，reduceBykey是个action操做,RDD中没有这个函数，是经过里面的隐式转换调用了PairRDDFunctions.scala中的reduceBykey()函数，里面的转换先不用管，由于涉及到shuffle操做，会有新的stage的生成，这里先略过：
def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)] = self.withScope { combineByKeyWithClassTag[V]((v: V) => v, func, func, partitioner) }
5. 最后主函数调用了wordCount.foreach(）进行告终果打印，这是一个action操做，有几个action就会提交几个job，直接去看代码：
def foreach(f: T => Unit): Unit = withScope { val cleanF = sc.clean(f) /*这里是执行了runJob,跟其余操做不同，这里会提交一个job*/ sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF)) }
跟进代码，里面调用了SparkContext.scala中的函数：
def runJob[T, U: ClassTag]( rdd: RDD[T], func: Iterator[T] => U, partitions: Seq[Int]): Array[U] = { //这里clean函数其实直接输出 val cleanedFunc = clean(func) runJob(rdd, (ctx: TaskContext, it: Iterator[T]) => cleanedFunc(it), partitions) }
跟进了好几层，最后看runJob干了啥：
def runJob[T, U: ClassTag]( rdd: RDD[T], func: (TaskContext, Iterator[T]) => U, partitions: Seq[Int], resultHandler: (Int, U) => Unit): Unit = { if (stopped.get()) { 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函数比较任务，这样就跟以前SparkContext源码剖析内容联系在一块儿了 dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get) progressBar.foreach(_.finishAll()) rdd.doCheckpoint() }
6.上面调用了DAGScheduler中的runJob函数，这个DAGScheduler是咱们在SparkContext初始化的时候执行的初始化，DAGSCheduler主要工做:建立Job，推断出每个Job的stage划分（DAG），跟踪RDD，实体化stage的输出，调度job，将stage以taskSet的形式提交给TaskScheduler的实现类，在集群上运运行，其中，TaskSet是一组能够当即运行的独立task，基于集群上已存在的数据，直接看下代码：
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 start = System.nanoTime /* 这里就一行比较重要，这里调用submitJob进行提交 */ val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties) ThreadUtils.awaitReady(waiter.completionFuture, Duration.Inf) // 下面这些就是任务结果的一些判断了 waiter.completionFuture.value.get match { case scala.util.Success(_) => logInfo("Job %d finished: %s, took %f s".format (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9)) case scala.util.Failure(exception) => logInfo("Job %d failed: %s, took %f s".format (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9)) // SPARK-8644: Include user stack trace in exceptions coming from DAGScheduler. val callerStackTrace = Thread.currentThread().getStackTrace.tail exception.setStackTrace(exception.getStackTrace ++ callerStackTrace) throw exception } }
下面就是调用了submitJob进行任务的提交，代码以下：
def submitJob[T, U]( rdd: RDD[T], func: (TaskContext, Iterator[T]) => U, partitions: Seq[Int], callSite: CallSite, resultHandler: (Int, U) => Unit, properties: Properties): JobWaiter[U] = { // 这里确认咱们提交的Partition存在 val maxPartitions = rdd.partitions.length partitions.find(p => p >= maxPartitions || p < 0).foreach { p => throw new IllegalArgumentException( "Attempting to access a non-existent partition: " + p + ". " + "Total number of partitions: " + maxPartitions) }
val jobId = nextJobId.getAndIncrement() if (partitions.size == 0) { // Return immediately if the job is running 0 tasks return new JobWaiter[U](this, jobId, 0, resultHandler) }
assert(partitions.size > 0) val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _] val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler) //这里会触发DAGSchedulerEventProcessLoop的JobSubmitted，他里面onReceive()函数 //接收消息进行处理，这里调用的是JobSubmitted，触发dagScheduler.handleJobSubmitted //函数进行处理 eventProcessLoop.post(JobSubmitted( jobId, rdd, func2, partitions.toArray, callSite, waiter, SerializationUtils.clone(properties))) waiter }
下面就是调用handleJobSubmitted()函数进行处理，它是DAGSchduler的job调度核心入口，代码以下：
private[scheduler] def handleJobSubmitted(jobId: Int, finalRDD: RDD[_], func: (TaskContext, Iterator[_]) => _, partitions: Array[Int], callSite: CallSite, listener: JobListener, properties: Properties) { // var finalStage: ResultStage = null try { //使用触发job的最后一个rdd，建立stage //当hdfs上的文件被删除的时候 stage可能建立失败 finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite) } catch { case e: Exception => logWarning("Creating new stage failed due to exception - job: " + jobId, e) listener.jobFailed(e) return } //经过finalStage创建立一个job, val job = new ActiveJob(jobId, finalStage, callSite, listener, properties) clearCacheLocs() logInfo("Got job %s (%s) with %d output partitions".format( job.jobId, callSite.shortForm, partitions.length)) logInfo("Final stage: " + finalStage + " (" + finalStage.name + ")") logInfo("Parents of final stage: " + finalStage.parents) logInfo("Missing parents: " + getMissingParentStages(finalStage))
val jobSubmissionTime = clock.getTimeMillis() //将job加入到activeJob缓存中 jobIdToActiveJob(jobId) = job activeJobs += job finalStage.setActiveJob(job) val stageIds = jobIdToStageIds(jobId).toArray val stageInfos = stageIds.flatMap(id => stageIdToStage.get(id).map(_.latestInfo)) listenerBus.post( SparkListenerJobStart(job.jobId, jobSubmissionTime, stageInfos, properties)) //提交finalStage，可是finalStage确定不会首先执行，它要先执行它的依赖stage submitStage(finalStage) }
7.最后调用了submitStage进行了finalStage的提交，finalStage确定不会首先执行，它要先执行它的依赖stage，这里面就涉及到了stage的换分了，代码以下：
/** Submits stage, but first recursively submits any missing parents. */ private def submitStage(stage: Stage) { val jobId = activeJobForStage(stage) if (jobId.isDefined) { logDebug("submitStage(" + stage + ")") if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) { //获取stage对应的父stage,返回List[Stage]按id排序 val missing = getMissingParentStages(stage).sortBy(_.id) logDebug("missing: " + missing) // 若是父stage为空，则调用submitMissingTasks 提交stage, if (missing.isEmpty) { logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents") submitMissingTasks(stage, jobId.get) } else { for (parent <- missing) { // 若是父stage不为空，则调用submitStage 提交父stage submitStage(parent) } //并将stage放入等待的队列中，先去执行父stage waitingStages += stage } } } else { abortStage(stage, "No active job for stage " + stage.id, None) } }
咱们看下getMissingParentStages()函数，如何进行stage划分的，代码以下：
//大致划分流程：遍历rdd的全部的依赖，若是是ShufDep，则经过getShuffleMapStage获取stage， // 并加入到missing队列中。若是是窄依赖的话，将放入waitingForVisit的栈中。 private def getMissingParentStages(stage: Stage): List[Stage] = { val missing = new HashSet[Stage] val visited = new HashSet[RDD[_]] // We are manually maintaining a stack here to prevent StackOverflowError // caused by recursively visiting val waitingForVisit = new Stack[RDD[_]] def visit(rdd: RDD[_]) { if (!visited(rdd)) { visited += rdd val rddHasUncachedPartitions = getCacheLocs(rdd).contains(Nil) if (rddHasUncachedPartitions) { for (dep <- rdd.dependencies) { dep match { //若是shufDep也就是咱们说的宽依赖 case shufDep: ShuffleDependency[_, _, _] => //宽依赖，则建立一个shuffleStage，即finalStage以前的stage是shuffle stage val mapStage = getOrCreateShuffleMapStage(shufDep, stage.firstJobId) if (!mapStage.isAvailable) { //加入到missing队列，返回 missing += mapStage } //若是narrowDep也就是咱们说的窄依赖 case narrowDep: NarrowDependency[_] => //加入等待队列中 waitingForVisit.push(narrowDep.rdd) } } } } } waitingForVisit.push(stage.rdd) while (waitingForVisit.nonEmpty) { // 若是是窄依赖，将rdd放入栈中 visit(waitingForVisit.pop()) } missing.toList }
submitStage()函数中若是父stage为空则，调用submitMissingTasks()函数进行提交，这个函数主要作了一下几件事：
a.首先获取stage中没有计算的partition；
b.经过 taskIdToLocations(id) 方法进行tasks运行最佳位置的肯定；
c.调用taskScheduler的submitTasks进行任务的提交。