spark源码阅读（一） 启动代码阅读

spark启动代码阅读：

spark使用一系列的shell脚本作为入口：其中bin目录下面是任务提交的脚本；sbin目录是master和worker启停相关的脚本。

而所有脚本最后都是通过调用bin/spark-class来实现对java（scala）代码的调用。

----------------------spark-class获取java参数分析 开始-------------------------------------

spark-class的代码处理流程：

  1. 调用org.apache.spark.launcher.Main并把传入的参数代入，以获取到具体的参数：

("$RUNNER" -cp "$LAUNCH_CLASSPATH" org.apache.spark.launcher.Main "[email protected]")将参数代入，即执行的是：

/usr/java/jdk/bin/java -cp /home/xxx/spark/lib/spark-assembly-1.5.2-hadoop2.5.0-cdh5.3.2.jar org.apache.spark.launcher.Main org.apache.spark.deploy.SparkSubmit --class com.xxx.xxxx.stat.core.Main --master spark://xxxx1:7077,xxxx2:7077 --executor-memory 2G --driver-memory 5G --total-executor-cores 10 /home/xxx/xxxxxxx/bigdata-xxxxxxx.jar com.xxx.xxxx.stat.xxx.XXXXJob 20180527 20180528

这行代码返回的是：

/usr/java/jdk/bin/java -cp /home/xxx/spark/libext/*:/home/xxx/spark/conf/:/home/xxx/spark/lib/spark-assembly-1.5.2-hadoop2.5.0-cdh5.3.2.jar:/home/xxx/spark/lib/datanucleus-api-jdo-3.2.6.jar:/home/xxx/spark/lib/datanucleus-core-3.2.10.jar:/home/xxx/spark/lib/datanucleus-rdbms-3.2.9.jar:/home/xxx/yarn/etc/hadoop -DLOG_LEVEL=INFO -DROLE_NAME=console -Xms5G -Xmx5G -Xss32M -XX:PermSize=128M -XX:MaxPermSize=512M org.apache.spark.deploy.SparkSubmit --master spark://xxxx1:7077,xxxx2:7077 --conf spark.driver.memory=5G --class com.xxx.xxxx.stat.core.Main --executor-memory 2G --total-executor-cores 10 /home/xxx/xxxxxxx/bigdata-xxxxxxx.jar com.xxx.xxxx.stat.xxx.XXXXJob 20180527 20180528

可以看出org.apache.spark.launcher.Main类的主要作用就是填充了最终执行的java命令的参数。包括classpath、java命令的堆栈参数等。下面分析下实现过程：

1.org.apache.spark.launcher.Main是一个单独的路径，不属于core的一部分。

2.org.apache.spark.launcher.Main这个类对于org.apache.spark.deploy.SparkSubmit这个的调用，创建了builder = new SparkSubmitCommandBuilder(args);用来规整sparksubmit命令的java参数。

3.对于其他的调用（应该主要是master和worker的调用），创建了builder = new SparkClassCommandBuilder(className, args);用于规整其他命令的java参数。

对于上面两种情况，都是先初始化，然后调用builder的buildCommand方法两步就生成了java代码的参数。

SparkSubmitCommandBuilder通过调用内部类class OptionParser extends SparkSubmitOptionParser 进行对参数的解析。

classpath主要在

buildCommand

buildSparkSubmitCommand

buildJavaCommand(在父类中)

在各种可能含有classpath的地方获取。

-Xms5G -Xmx5G这两个参数是在buildSparkSubmitCommand中的通过参数中的SPARK_DRIVER_MEMORY(spark.driver.memory)解析获取到的

-Xss32M -XX:PermSize=128M -XX:MaxPermSize=512M则是通过addPermGenSizeOpt(cmd);在配置文件中的spark.driver.extraJavaOptions这个配置项（DRIVER_EXTRA_JAVA_OPTIONS = "spark.driver.extraJavaOptions";）解析获取到的。

----------------------------spark-class获取java参数分析 结束----------------------------------------------------------

先分析下目前使用较多的bin/spark-submit(spark任务提交，driver进程启动)、sbin/start-master.sh（后台启动，master进程启动）、sbin/start-slave.sh(后台启动，worker进程启动)，启动代码都在spark-1.5.2\core\src\main\scala\org\apache\spark\deploy：

  1. spark-submit

通过spark-class调用的类名：org.apache.spark.deploy.SparkSubmit

下面来分析，是怎么把driver启动起来的，分析类SparkSubmit.scala

/usr/java/jdk/bin/java -cp /home/xxx/spark/libext/*:/home/xxx/spark/conf/:/home/xxx/spark/lib/spark-assembly-1.5.2-hadoop2.5.0-cdh5.3.2.jar:/home/xxx/spark/lib/datanucleus-api-jdo-3.2.6.jar:/home/xxx/spark/lib/datanucleus-core-3.2.10.jar:/home/xxx/spark/lib/datanucleus-rdbms-3.2.9.jar:/home/xxx/yarn/etc/hadoop -DLOG_LEVEL=INFO -DROLE_NAME=console -Xms5G -Xmx5G -Xss32M -XX:PermSize=128M -XX:MaxPermSize=512M org.apache.spark.deploy.SparkSubmit --master spark://xxxx1:7077,xxxx2:7077 --conf spark.driver.memory=5G --class com.xxx.xxxx.stat.core.Main --executor-memory 2G --total-executor-cores 10 /home/xxx/xxxxxxx/bigdata-xxxxxxx.jar com.xxx.xxxx.stat.xxx.XXXXJob 20180527 20180528

传入的参数主要是：

--conf spark.driver.memory=5G

--class com.xxx.xxxx.stat.core.ExcuteMain

--executor-memory 2G

--total-executor-cores 10

/home/xxx/xxxxxxx/bigdata-xxxxxxx.jar

com.xxx.xxxx.stat.xxx.XXXXJob

20180527

20180528

SparkSubmit的Main函数，通过val appArgs = new SparkSubmitArguments(args)获取到进一步解析的参数，然后调用submit(appArgs)，就实现了提交。

SparkSubmitArguments类，先调用上面org.apache.spark.launcher.Main调用过的org.apache.spark.launcher.SparkSubmitOptionParser的parse来对参数进行解析，然后调用loadEnvironmentArguments对环境中可能配置的参数进行解析或者赋予默认值。最后给action参数赋予SUBMIT的默认值：

action = Option(action).getOrElse(SUBMIT)

下面看submit的方法的处理过程：

  1. val (childArgs, childClasspath, sysProps, childMainClass) = prepareSubmitEnvironment(args)

a.定义driver代码类childArgs、childClasspath、sysProps、childMainClass。可以这样理解：submit提交了很多信息，比如使用的核数（对应的executor的个数）、每个核用的内存数、执行的driver代码（driver代码也视为一种提交的内容）

b.定义集群管理clusterManager，根据master的前缀进行区分。yarn,spark,mesos,local

c.定义提交模式deployMode，CLIENT则driver在当前机器；CLUSTER则使用某个worker作为driver。

d.下面的代码就是对clusterManager、deployMode、python(R)组合的特殊情况处理。我们重点关注standalone模式。

e.将各个参数填入options变量中。

f.对于if (deployMode == CLIENT) { 填充四个参数。 直接将childMainClass = args.mainClass填充，在sparkSubmit中直接被runMain调用执性。

f.对于isStandaloneCluster模式（standalone和cluster模式），区分legacy和rest两种事项方式来启动一个client来执性dirver

rest方式将org.apache.spark.deploy.rest.RestSubmissionClient填充到childMainClass，

legacy方式：org.apache.spark.deploy.Client填充到childMainClass

在sparksubmit中执行上面的类，而将args.mainClass作为参数传给上面的类。

g.对于cluster模式忽略spark.driver.host参数。

h.返回四个参数

四个参数的解释：

  • This returns a 4-tuple:
  • (1) the arguments for the child process,
  • (2) a list of classpath entries for the child,
  • (3) a map of system properties, and
  • (4) the main class for the child
  1. 调用doRunMain

runMain

调用提交的childMainClass，

mainClass = Utils.classForName(childMainClass)

val mainMethod = mainClass.getMethod("main", new ArrayString.getClass)

mainMethod.invoke(null, childArgs.toArray)

  1. start-master.sh(中间调用了spark-daemon.sh)

通过spark-class调用的类名：org.apache.spark.deploy.master.Master

调用时带的参数：

  1. start-slave.sh(中间调用了spark-daemon.sh)

通过spark-class调用的类名：org.apache.spark.deploy.worker.Worker

调用时带的参数：

原文地址：http://blog.51cto.com/11091005/2121281

时间： 2024-10-13 03:10:39

spark源码阅读（一） 启动代码阅读的相关文章

[Apache Spark源码阅读]天堂之门——SparkContext解析

稍微了解Spark源码的人应该都知道SparkContext,作为整个Project的程序入口,其重要性不言而喻,许多大牛也在源码分析的文章中对其做了很多相关的深入分析和解读.这里,结合自己前段时间的阅读体会,与大家共同讨论学习一下Spark的入口对象—天堂之门—SparkContex. SparkContex位于项目的源码路径\spark-master\core\src\main\scala\org\apache\spark\SparkContext.scala中,源文件包含Classs Sp

【Spark】配置Spark源码阅读环境

Scala构建工具(SBT)的使用 SBT介绍 SBT是Simple Build Tool的简称,如果读者使用过Maven,那么可以简单将SBT看做是Scala世界的Maven,虽然二者各有优劣,但完成的工作基本是类似的. 虽然Maven同样可以管理Scala项目的依赖并进行构建,但SBT的某些特性却让人如此着迷,比如: 使用Scala作为DSL来定义build文件(one language rules them all); 通过触发执行(trigger execution)特性支持持续的编译与

Spark源码阅读笔记之Broadcast(一)

Spark源码阅读笔记之Broadcast(一) Spark会序列化在各个任务上使用到的变量,然后传递到Executor中,由于Executor中得到的只是变量的拷贝,因此对变量的改变只在该Executor有效.序列化后的任务的大小是有限制的(由spark.akka.frameSize决定,值为其减去200K,默认为10M-200K),spark会进行检查,超出该限制的任务会被抛弃.因此,对于需要共享比较大的数据时,需要使用Broadcast. Spark实现了两种传输Broadcast的机制:

第2课 Scala面向对象彻底精通及Spark源码SparkContext，RDD阅读总结

第2课:Scala面向对象彻底精通及Spark源码阅读本期内容:1 Scala中的类.object实战详解 2 Scala中的抽象类.接口实战详解 3 综合案例及Spark源码解析 一:定义类class HiScala{private var name = "Spark" def sayName(){println(name)}def getName = name} Scala中,变量与类中的方法是同等级的,可以直接赋值给方法. scala中的get与set与Java中的get,set

第3课 Scala函数式编程彻底精通及Spark源码阅读笔记

本课内容: 1:scala中函数式编程彻底详解 2:Spark源码中的scala函数式编程 3:案例和作业 函数式编程开始: def fun1(name: String){ println(name) } //将函数名赋值给一个变量,那么这个变量就是一个函数了. val fun1_v = fun1_ 访问 fun1_v("Scala") 结果:Scala 匿名函数:参数名称用 => 指向函数体 val fun2=(content: String) => println(co

《STL源码剖析》---stl_iterator.h阅读笔记

STL设计的中心思想是将容器(container)和算法(algorithm)分开,迭代器是容器(container)和算法(algorithm)之间的桥梁. 迭代器可以如下定义:提供一种方法,能够依序寻访某个容器内的所有元素,而又无需暴露该容器的内部表达方式. 在阅读代码之前,要先了解一个新概念:Traits编程技法 template <class T> struct MyIter { typedef T value_type //内嵌型别声明 T *ptr; MyIter(T *p = 0

spark源码分析之Executor启动与任务提交篇

任务提交流程 概述 在阐明了Spark的Master的启动流程与Worker启动流程.接下继续执行的就是Worker上的Executor进程了,本文继续分析整个Executor的启动与任务提交流程 Spark-submit 提交一个任务到集群通过的是Spark-submit 通过启动脚本的方式启动它的主类,这里以WordCount为例子 spark-submit --class cn.itcast.spark.WordCount bin/spark-clas -> org.apache.spar

《STL源码剖析》---stl_hashtable.h阅读笔记

在前面介绍的RB-tree红黑树中,可以看出红黑树的插入.查找.删除的平均时间复杂度为O(nlogn).但这是基于一个假设:输入数据具有随机性.而哈希表/散列表hash table在插入.删除.查找上具有"平均常数时间复杂度"O(1):且不依赖输入数据的随机性. hash table的实现有线性探测.二次探测.二次散列等实现,SGI的STL是采用开链法(separate chaining)来实现的.大概原理就是在hash table的每一项都是个指针(指向一个链表),叫做bucket.

《STL源码剖析》---stl_alloc.h阅读笔记

这一节是讲空间的配置与释放,但不涉及对象的构造和析构,只是讲解对象构造前空前的申请以及对象析构后空间怎么释放. SGI版本的STL对空间的的申请和释放做了如下考虑: 1.向堆申请空间 2.考虑了多线程.但是这节目的只是讲解空间配置与释放,因此忽略了多线程,集中学习空间的申请和释放. 3.内存不足时的应变措施 4.考虑到了内存碎片的问题.多次申请释放小块内存可能会造成内存碎片. 在C++中,内存的申请和释放是通过operator new函数和operator delete函数,这两个函数相当于C语