Spark集群Job，Task 的具体运行原理

一：Ｓｐａｒｋ集群部署

二：Ｊｏｂ提交解密

三：Ｊｏｂ生成和接受

四：Ｔａｓｋ的运行

五：再论ｓｈｕｆｆｌｅ

１，从ｓｐａｒｋ　Ｒｕｎｔｉｍｅ　的角度讲来讲有５大核心对象：Ｍａｓｔｅｒ　，　Ｗｏｒｋｅｒ　，　Ｅｘｅｃｕｔｏｒ　，Ｄｒｉｖｅｒ　，　ＣｏａｒｓｅＧｒａｉｎｅｄＥｘｅｃｕｔｏｒｂａｃｋｅｄ　；

２，Ｓｐａｒｋ　在做分布式集群系统的设计的时候，最大化功能的独立，模块化封装具体的独立的对象，强内聚低耦合　　　（耦合性也称块间联系，指软件系统结构中各模块间相互联系紧密程度的一种度量。模块之间联系越紧密，其耦合性就越强，模块的独立性则越差。模块间耦合高低取决于模块间接口的复杂性、调用的方式及传递的信息。内聚性又称块内联系。指模块的功能强度的度量，即一个模块内部各个元素彼此结合的紧密程度的度量。若一个模块内各元素（语名之间、程序段之间）联系的越紧密，则它的内聚性就越高。）



３，当Ｄｒｉｖｅｒ中的ｓｐａｒｋＣｏｎｔｅｘｔ　初始化的时候会提交程序给Ｍａｓｔｅｒ，Ｍａｓｔｅｒ如果接受该程序在ｓｐａｒｋ中运行的话，就会为当前程序分配ＡｐｐＩＤ　，同时分配计算资源，需要特备注意的是：　Ｍａｓｔｅｒ是根据当前程序的配置信息来给集群中的Ｗｏｒｋｅｒ发指令来分配具体的计算资源。但是，Ｍａｓｔｅｒ发指令后并不关心具体的计算资源是否已经分配，转过来说，Ｍａｓｔｅｒ发出指令后就记录了分配的资源，以后客户端再次提交其他程序的话就不能使用该资源啦，其弊端是可能会导致其他要提交的程序无法分配到本来应该可以分配到的计算资源。最终优势在ｓｐａｒｋ分布式系统功能弱耦合的基础上最快的运行系统（否则如果Ｍａｓｔｅｒ要等到计算资源最终分配成功后才通知Ｄｒｉｖｅｒ的话，会造成Ｄｒｉｖｅｒ的阻塞，不能够最大化的并行计算资源的利用率）　　（低耦合　：　不关心指令发送成功还是失败）　　（快是对Ｄｒｉｖｅｒ　而言）

补充说明的是：　Ｓｐａｒｋ默认程序是排队的，Ｓｐａｒｋ默认的情况下由于集群中一般都只有一个Ａｐｐｌｉｃａｔｉｏｎ在运行，所有Ｍａｓｔｅｒ分配计算资源策略就没有那么明显啦）

二　：　Ｊｏｂ提交过程　源码解密

１，一个非常重要的技巧通过在Ｓｐａｒｋ－ｓｈｅｌｌ　中运行一个Ｊｏｂ来了解Ｊｏｂ提交的过程，然后再次用源码验证。

这个过程　　：　

　　ｓｃ．ｔｅｘｔＦｉｌｅ（＂ｌｉｂｒａｒｙ／ｄａｔａ１＂）．ｆｌａｔＭａｐ（＿．ｓｐｌｉｔ（＂＂））．ｍａｐ（ｗｏｒｄ　＝＞　（ｗｏｒｄ，１））．ｒｅｄｕｃｅＢｙＫｅｙ（＿＋＿）ｓａｖｅＡｓＴｅｘｔＦｉｌｅ（＂／ｌｉｂｒａｒｙ／ｄａｔａ２＂）

２，在Ｓｐａｒｋ中所有的Ａｃｔｉｏｎ都会触发一个至少一个Ｊｏｂ，在上述代码中通过ｓａｖａＡｓＴｅｘｔＦｉｌｅ来触发Ｊｏｂ的

３.ＳｐａｒｋＣｏｎｔｅｘｔ　在实例化的时候会构造ＳｐａｒｋＤｅｐｌｏｙＳｈｅｄｕｌｅｒＢａｃｋｅｎｄ（ｄｅｐｌｏｙ　：　配置，部署），ＤＡＧＳｃｈｅｄｕｌｅｒ，ＴａｓｋＳｈｅｄｕｌｅｒＩｍｐｌ（Ｉｍｐｌ　： 接口），MapOutputTrackerMaster（Tracker ： 追踪）等对象：

（1）SparkDeploySchedulerBackend负责集群计算资源的管理和调度。

（2）DAGScheduler ： 负责高层调度（例如： Job中stage的划分，数据本地性等内容）

（3）TaskShedulerImpl : 负责具体stage内部的底层调度（例如： 每个Task的调度 ，Task容错等等）

（4）MapOutputTrackerMaster： 负责shuffle中数据的输出和读取的管理。

4，TaskSchedulerImpl内部的调度：



三：Ｔａｓｋ　的运行解密：

１，Ｔａｓｋ运行在Ｅｘｅｃｕｔｏｒ中，而Ｅｘｅｃｕｔｏｒ又是位于ＣｏａｒｓｅＧｒａｉｎｅｄＥｘｅｃｕｔｏｒＢａｃｋｅｎｄ中的且ＣｏａｒｓｅＧｒａｉｎｅｄＥｘｅｃｕｔｏｒＢａｃｋｅｎｄ和Ｅｘｅｃｕｔｏｒ是一一对应的：

２，单ＣｏａｒｓｅＧｒａｉｎｅｄＥｘｅｃｕｔｏｒＢａｃｋｅｎｄ接受到ＴａｓｋＳｅｔＭａｎａｇｅｒ发过来的ＬａｕｎｃｈＴａｓｋ的消息后会反序列化ＴａｓｋＤｅｓｃｒｉｐｔｉｏｎ，然后使用ＣｏａｒｓｅＧｒａｉｎｅｄＥｘｅｃｕｔｏｒＢａｃｋｅｎｄ中唯一的Ｅｘｅｃｕｔｏｒ来执行任务

case LaunchTask(data) =>

if (executor == null) {

logError(“Received LaunchTask command but executor was null”)

System.exit(1)

} else {

val taskDesc = ser.deserializeTaskDescription

logInfo(“Got assigned task ” + taskDesc.taskId)

executor.launchTask(this, taskId = taskDesc.taskId, attemptNumber = taskDesc.attemptNumber,

taskDesc.name, taskDesc.serializedTask)

}

发消息要么是ｃａｓｅ　ｃｌａｓｓ　或者　ｃａｓｅ　ｏｂｊｅｃｔ（是唯一的）每次生成类的事例

本博客内容来自于 ：　简介： 王家林：DT大数据梦工厂创始人和首席专家. 联系邮箱18610086859@126.com 电话：18610086859 QQ:1740415547 微信号：18610086859