spark任务运行的源码分析
在整个spark任务的编写、提交、执行分三个部分:
① 编写程序和提交任务到集群中
②sparkContext的初始化
③触发action算子中的runJob方法,执行任务
创新互联主要从事做网站、
成都网站设计、网页设计、企业做网站、公司建网站等业务。立足成都服务京山,10年网站建设经验,价格优惠、服务专业,欢迎来电咨询建站服务:18982081108
(1)编程程序并提交到集群:
①编程spark程序的代码
②打成jar包到集群中运行
③使用spark-submit命令提交任务
在提交任务时,需要指定 --class 程序的入口(有main方法的类),
1) spark-submit --class xxx
2) ${SPARK_HOME}/bin/spark-class org.apache.spark.deploy.SparkSubmit $@
3) org.apache.spark.launcher.Main
submit(appArgs, uninitLog)
doRunMain()
runMain(childArgs, childClasspath, sparkConf, childMainClass, args.verbose)
childMainClass:…./.WordCount (自己编写的代码的主类)
mainClass = Utils.classForName(childMainClass)
val app: SparkApplication = if() {} else {new JavaMainApplication(mainClass)}
app.start(childArgs.toArray, sparkConf) // 通过反射调用mainClass执行
// 到此为止,相当于调用了我们的自己编写的任务类的main方法执行了。!!!
val mainMethod = klass.getMethod("main", new ArrayString.getClass)
mainMethod.invoke(null, args)
④开始执行自己编写的代码
(2)初始化sparkContext:
当自己编写的程序运行到:new SparkContext()时,就开始了精妙而细致的sparkContext的初始化。
sparkContext的相关介绍:sparkContext是用户通往spark集群的唯一入口,可以用来在spark集群中创建RDD、累加器和广播变量。sparkContext也是整个spark应用程序的一个至关重要的对象,是整个应用程序运行调度的核心(不是资源调度的核心)。在初始化sparkContext时,同时的会初始化DAGScheduler、TaskScheduler和SchedulerBackend,这些至关重要的对象。
sparkContext的构建过程:
1)Driver端执行的代码:
初始化 TaskScheduler
初始化 SchedulerBackend
初始化 DAGScheduler
2)worker和master端执行的代码:
driver向master注册申请资源。
Worker负责启动executor。
(3)触发action算子中的runJob方法:
spark任务运行总结:
- 将编写的程序打成jar包
- 调用spark-submit提交任务到集群上运行
- 运行sparkSubmit 的main方法,在这个方法中通过反射的方式创建我们编写的主类的实例对象,然后调用该对象的main方法,开始执行我们编写的代码
- 当代码运行到new SparkContext对象的的时候,就开始了复杂和精致的sparkContext对象的初始化
- 在初始化SparkContext对象的时候,会创建两个特别重要的对象,分别是:DAGScheduler 和 TaskScheduler,其中【DAGScheduler 的作用】将RDD的依赖切成一个一个的stage,然后stage作为taskSet提交给Taskscheduler。
- 在构建TaskScheduler的同时,会创建两个非常重要的对象,分别是 DriverActor 和 ClientActor,DriverActor负责接收executor的反向注册,将任务提交给executor运行,clientActor是负责向master注册并提交任务
- 当clientActor启动时,会将用户提交的任务相关的参数分装到applicationDescription对象中去,然后提交给master进行任务注册
- 当master接收到clientActor提交的任务请求时,会将请求的参数进行分析,并封装成application,然后将其持久化,然后将其加入到任务队列waitingApps中。
- 当轮到我们提交任务的时候,就开始执行schedule(),进行任务资源的调度
- worker接收到master发送来的launchExecutor 时,会将其解压并封装到ExecutorRunner中,然后调用这个对象的start方法,启动executor
- executor启动后会向driver反向注册
- driver会发送注册成功信息,给executor
- executor接收到driver actor注册成功信息后,就会创建一个线程池,用于执行driveractor发送过来的任务
- 当属于这个任务的所有的 Executor 启动并反向注册成功后,就意味着运行这个任务的 环境已经准备好了,driver 会结束 SparkContext 对象的初始化,也就意味着 new SparkContext 这句代码运行完成
- 当sparkContext初始化完成之后,就会继续运行我们的代码,直到运行到action算子时,也就意味着触发了一个job的提交
- driver 会将这个 job 提交给 DAGScheduler
- DAGScheduler将接收到的job,从最后一个算子开始推导,将DAG根据依赖关系划分成为一个个stage,然后将stage封装成一个taskSet,并将taskSet中的task提交给taskScheduler
- taskScheduler接收到DAGScheduler发送过来的task,会拿到一个序列化器,对task进行序列化,然后将序列化好的task封装到launchTask中,然后将launchTask发送给指定的executor中运行
- executor接收到了DriverActor 发送过来的launchTask 时,会拿到一个反序列化器,对launchTask 进行反序列化,封装到一个TaskRunner 中,然后从executor这个线程池中获取一个线程,将反序列化好的任务中的算子作用在RDD对应的分区上。
- 最终当所有的task任务完成之后,整个application执行完成,关闭sparkContext对象。
另外有需要云服务器可以了解下创新互联scvps.cn,海内外云服务器15元起步,三天无理由+7*72小时售后在线,公司持有idc许可证,提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案,具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势,专为企业上云打造定制,能够满足用户丰富、多元化的应用场景需求。
名称栏目:spark任务运行过程的源码分析-创新互联
网站路径:
http://njwzjz.com/article/coochg.html