Spark源码分析

各个组件介绍

后面补充。。。。

StandAlone模式

在StandAlone模式的start-all的shell启动脚本下，在当前机器执行了JAVA_HOME/bin/java -cp ....Master和在配置的slave的机器中执行 JAVA_HOME/bin/java -cp ....Worker.这两种进程在启动后通过netty进行rpc通信。

Master的启动

首先创建一个RpcEnv对象，负责管理所有通信逻辑，核心代码为val rpcEnv: RpcEnv = RpcEnv.create(SYstem_NAME, host, port, conf, securityMgr),创建后启动NettyRpcEnv Utils.startServiceOnPort(config.port, startNettyRpcEnv, sparkConf, config.name)._1
接着创建一个Master的EndPoint对象val masterEndpoint: RpcEndpointRef = rpcEnv.setupEndpoint(ENDPOINT_NAME,new Master(rpcEnv, rpcEnv.address, webUiPort, securityMgr, conf))
该EndPoint对象有constructor -> onStart -> receive* -> onStop这样一个生命周期

Worker的启动

Worker以类似的情形创建RpcEnv和Worker的EndPoint
Worker作为一个EndPoint，其同样有constructor -> onStart -> receive* -> onStop
创建一个masterEndpointRef对象val masterEndpoint: RpcEndpointRef = rpcEnv.setupEndpoint(ENDPOINT_NAME,new Master(rpcEnv, rpcEnv.address, webUiPort, securityMgr, conf))

Master和Worker的通信

在Worker的Onstart阶段（方法）向Master注册。 registerWithMaster() -->通过Master的引用通信masterEndpoint.ask[RegisterWorkerResponse](RegisterWorker(workerId, host, port, self, cores, memory, workerWebUiUrl)).onComplete-->ask请求的Master接收再回复，调用receiveAndReply-->成功回复context.reply(RegisteredWorker(self, masterWebUiUrl))-->worker在收到注册成功后，开始发送心跳forwordMessageScheduler.scheduleAtFixedrate(new Runnable
之后进入receive阶段worker不断发送心跳 sendToMaster(Heartbeat(workerId, self))，Master不断记录心跳workerInfo.lastHeartbeat = System.currentTimeMillis()

Yarn On Spark模式

Yarn Cluster

总体流程图

执行Submit中的main方法, 反射调用client的main方法
向rm提交bin/java applicationMaster
rm寻找一台NN启动applicationMaster
NN节点启动Driver进程
Driver向RM申请资源
Driver获得资源后向一个NN发送指令，bin/java CoarseGrainedExecutorBackend
启动ExecutorBackend，这个进程启动后向Driver进行注册
Driver返回注册成功的信息
ExectureBack创建一个Execture对象
Driver向Execture分配任务

创建Sparksubmit创建

从SparkSubmit类的main方法开始，调用submit(appArgs)
准备运行环境prepareSubmitEnvironment
调用doRunMain，之后再调用runMain()方法。
使用反射的方式加载 childMainClass = "org.apache.spark.deploy.yarn.Client"
执行Client的main方法mainMethod.invoke(null, childArgs.toArray)
调用new Client(args, sparkConf).run(),创建一个Client对象并运行它的run方法

创建Driver

Client 对象中的run方法会启动一个yarn的客户端，与yarn集群通信
通过客户端设置正确的上下文对象来启动 ApplicationMaster val containerContext = createContainerLaunchContext(newAppResponse)
源码进入ApplicationMaster了，在其main方法中创建了一个master对象，并调用这个对象的run()方法，在run方法中调用了runDriver(securityMgr),runDriver是一个子线程。
这个线程像RM注册并且申请资源 allocator.allocateResources()
申请到资源之后 prepareCommand（）依旧是准备一个command执行。/bin/java org.apache.spark.executor.

创建ExecutorBacken

启动ExecutorBackend, 并向driver注册. Driver.ask[Boolean](RegisterExecutor(executorId, self, hostname, cores, extractLogUrls))}(ThreadUtils.sameThread).onComplete （on start）
注册成功后, ExecutorBackend会创建一个Executor对象.executor = new Executor(executorId, hostname, env, userClasspath, isLocal = false) //创建计算对象 (on recieve)
Driver会给ExecutorBackend分配任务, 并监控任务的执行.

Yarn Client模式

Shuffle分析

碰到reduceByKey不一定shuffle。

gdgylpc 发布了118 篇原创文章 · 获赞 5 · 访问量 7143 私信关注

15.Spark源码分析