spark-Master注册机制

Posted 2020-11-16 parent-absent-son

一：worker注册机制（worker启动之后）

1- worker在启动之后，就会主动向master进行注册；

2- Master将dead状态的worker过滤掉；将unknown状态的worker信息更新；

3- Master将worker将入到内存缓存中（HashMap）；

4- 持久化引擎将worker信息进行持久化（zookeeper和文件系统中）；

5- 调用scheduler（）方法，进行资源调度；

源代码见：Master.scala的RegisterWorker（）方法。

 

二：Driver注册（在spark-submit的时候进行注册）；

1- 将diver信息放到内存缓存中（HashMap）;

2- 加入等待调度队列（ArrayBuffer）；

3- 持久化引擎将driver信息进行持久化；

4- 调用scheduler（）方法，进行资源调度；

源码见：Master.scala的ReceiverAndReply（）方法进行监控；

             Client.scala的RegisterSubmitDriver()方法进行注册；

 

三： Application注册（执行sparkContext的时候，底层会触发其调度）

1- 将application信息放到缓存中（HashMap）；

2- 将application加入到等待调度的application队列中；

3- 持久化引擎将application信息进行持久化；

4- 调用scheduler（）方法，进行资源调度；

源代码见：Master.scala的RegisterApplication()方法：

    case RegisterApplication(description, driver) =>
      // TODO Prevent repeated registrations from some driver
       //如果是standby状态，则不是active的master，不执行任何操作
      if (state == RecoveryState.STANDBY) {
        // ignore, don‘t send response
      //否则，执行以下行为
      } else {
        logInfo("Registering app " + description.name)
//创建app信息类
        val app = createApplication(description, driver)
//注册application，将application信息放到缓存中，并且加入到等待调度的队列中
        registerApplication(app)
        logInfo("Registered app " + description.name + " with ID " + app.id)
//使用持久化殷勤，将application进行持久化
        persistenceEngine.addApplication(app)
//反向向sparkDeploySchedulerBackend的appClient的ClientActor发送消息，也就是RegisteredApplication
        driver.send(RegisteredApplication(app.id, self))
//执行scheduler，进行资源调度
        schedule()
      }

 扩展：spark从入门到精通，中华石杉，第46讲；

 

四： driver状态改变

1- 判断driver的状态；

2- 将driver从内存中（hashMap）移除；  将持久化信息进行移除；  移除该driver相关的worker信息；   调用scheduler方法；

源码见Master.scala的DriverStateChanged()方法。

 

五：Excetor状态改变

1- 找到状态改变的Excetor；

2- 向driver发送excutorUpdated消息；

3- 如果Excetor执行完毕，将worker信息进行移除；

   如果没有执行完毕，则进行重试功能，默认次数是10次；

   如果超过10次，将application信息进行移除操作；

源码见：Master.scala的ExcutorChanged方法。

 

拓展阅读见：中华石杉-spark从入门到精通第47讲