189 8069 5689
如何进行Apache Spark源码分析Job的提交与运行 ，相信很多没有经验的人对此束手无策，为此本文总结了问题出现的原因和解决方法，通过这篇文章希望你能解决这个问题。
创新互联成立十载来，这条路我们正越走越好，积累了技术与客户资源，形成了良好的口碑。为客户提供网站设计制作、成都网站建设、网站策划、网页设计、域名与空间、网络营销、VI设计、网站改版、漏洞修补等服务。网站是否美观、功能强大、用户体验好、性价比高、打开快等等，这些对于网站建设都非常重要，创新互联通过对建站技术性的掌握、对创意设计的研究为客户提供一站式互联网解决方案，携手广大客户，共同发展进步。
下面以wordCount为例，详细说明spark创建和运行job的过程，重点是在进程及线程的创建。
实验环境搭建
在进行后续操作前，确保下列条件已满足。
1. 下载spark binary 0.9.1
2. 安装scala
3. 安装sbt
4. 安装java
启动spark-shell单机模式运行，即local模式
local模式运行非常简单，只要运行以下命令即可，假设当前目录是$SPARK_HOME
MASTER=local bin/spark-shell
"MASTER=local"就是表明当前运行在单机模式
local cluster方式运行
localcluster模式是一种伪cluster模式，在单机环境下模拟standalone的集群，启动顺序分别如下
1. 启动master
2. 启动worker
3. 启动spark-shell
master$SPARK_HOME/sbin/start-master.sh
注意运行时的输出，日志默认保存在$SPARK_HOME/logs目录。
master主要是运行类 org.apache.spark.deploy.master.Master，在8080端口启动监听，日志如下图所示
修改配置
1. 进入$SPARK_HOME/conf目录
2. 将spark-env.sh.template重命名为spark-env.sh
3. 修改spark-env.sh，添加如下内容
export SPARK_MASTER_IP=localhostexport SPARK_LOCAL_IP=localhost运行workerbin/spark-class org.apache.spark.deploy.worker.Worker spark://localhost:7077 -i 127.0.0.1 -c 1 -m 512M
worker启动完成，连接到master。打开maser的webui可以看到连接上来的worker. Master WEb UI的监听地址是http://localhost:8080
启动spark-shellMASTER=spark://localhost:7077 bin/spark-shell
如果一切顺利，将看到下面的提示信息。
Created spark context..Spark context available as sc.
可以用浏览器打开localhost:4040来查看如下内容
1. stages
2. storage
3. environment
4. executors
wordcount
上述环境准备妥当之后，我们在sparkshell中运行一下最简单的例子，在spark-shell中输入如下代码
scala>sc.textFile("README.md").filter(_.contains("Spark")).count
上述代码统计在README.md中含有Spark的行数有多少
部署过程详解
Spark布置环境中组件构成如下图所示。
Driver Program 简要来说在spark-shell中输入的wordcount语句对应于上图的Driver Program.
Cluster Manager 就是对应于上面提到的master，主要起到deploy management的作用
Worker Node 与Master相比，这是slave node。上面运行各个executor，executor可以对应于线程。executor处理两种基本的业务逻辑，一种就是driver programme,另一种就是job在提交之后拆分成各个stage，每个stage可以运行一到多个task
Notes: 在集群(cluster)方式下, Cluster Manager运行在一个jvm进程之中，而worker运行在另一个jvm进程中。在local cluster中，这些jvm进程都在同一台机器中，如果是真正的standalone或Mesos及Yarn集群，worker与master或分布于不同的主机之上。
JOB的生成和运行
job生成的简单流程如下
1. 首先应用程序创建SparkContext的实例，如实例为sc
2. 利用SparkContext的实例来创建生成RDD
3. 经过一连串的transformation操作，原始的RDD转换成为其它类型的RDD
4. 当action作用于转换之后RDD时，会调用SparkContext的runJob方法
5. sc.runJob的调用是后面一连串反应的起点，关键性的跃变就发生在此处
调用路径大致如下
1. sc.runJob->dagScheduler.runJob->submitJob
2. DAGScheduler::submitJob会创建JobSummitted的event发送给内嵌类eventProcessActor
3. eventProcessActor在接收到JobSubmmitted之后调用processEvent处理函数
4. job到stage的转换，生成finalStage并提交运行，关键是调用submitStage
5. 在submitStage中会计算stage之间的依赖关系，依赖关系分为宽依赖和窄依赖两种
6. 如果计算中发现当前的stage没有任何依赖或者所有的依赖都已经准备完毕，则提交task
7. 提交task是调用函数submitMissingTasks来完成
8. task真正运行在哪个worker上面是由TaskScheduler来管理，也就是上面的submitMissingTasks会调用TaskScheduler::submitTasks
9. TaskSchedulerImpl中会根据Spark的当前运行模式来创建相应的backend,如果是在单机运行则创建LocalBackend
10. LocalBackend收到TaskSchedulerImpl传递进来的ReceiveOffers事件
11. receiveOffers->executor.launchTask->TaskRunner.run
代码片段executor.lauchTask
def launchTask(context: ExecutorBackend, taskId: Long, serializedTask: ByteBuffer) { val tr = new TaskRunner(context, taskId, serializedTask) runningTasks.put(taskId, tr) threadPool.execute(tr) }
说了这么一大通，也就是讲最终的逻辑处理切切实实是发生在TaskRunner这么一个executor之内。
运算结果是包装成为MapStatus然后通过一系列的内部消息传递，反馈到DAGScheduler，这一个消息传递路径不是过于复杂。
看完上述内容，你们掌握如何进行Apache Spark源码分析Job的提交与运行 的方法了吗？如果还想学到更多技能或想了解更多相关内容，欢迎关注创新互联行业资讯频道，感谢各位的阅读！