温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

spark任务运行过程的源码分析

发布时间:2020-06-09 21:15:15 来源:网络 阅读:1261 作者:原生zzy 栏目:大数据

spark任务运行的源码分析

在整个spark任务的编写、提交、执行分三个部分:
① 编写程序和提交任务到集群中
②sparkContext的初始化
③触发action算子中的runJob方法,执行任务

(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的构建过程
spark任务运行过程的源码分析

1)Driver端执行的代码:

初始化 TaskScheduler
 初始化 SchedulerBackend
 初始化 DAGScheduler

spark任务运行过程的源码分析

2)worker和master端执行的代码:

driver向master注册申请资源。
  Worker负责启动executor。

spark任务运行过程的源码分析

(3)触发action算子中的runJob方法:

spark任务运行过程的源码分析

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对象。
向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI