spark——blockmanager原理与源码分析

89 阅读 0 评论 59 点赞

我是靠谱客的博主正直绿草，这篇文章主要介绍spark——blockmanager原理与源码分析，现在分享给大家，希望可以做个参考。

1. BlockManager定义
  1. BlockManager是Spark的分布式存储系统，与我们平常说的分布式存储系统是有区别的，区别就是这个分布式存储系统只会管理Block块数据，它运行在所有节点上。BlockManager的结构是Maser-Slave架构，Master就是Driver上的BlockManagerMaster，Slave就是每个Executor上的BlockManager。BlockManagerMaster负责接受Executor上的BlockManager的注册以及管理BlockManager的元数据信息
  2. 有需要可以联系我2317384986 yxxy1717
  3. BlockManager原理
2. 1. 从上边的定义我们已经得知，BlockManager是分布式的，运行在各个节点上的。从BlockManager的创建过程来看，其实Block是运行在Driver和每个Executor的。因为在创建SparkContext的时候，会调用SparkEnv.blockManager.initialize方法实例化BlockManager对象，在创建Executor对象的时候也会创建BlockManager。
  2. 在初始化BlockManager的时候，第一步会初始化BlockTransferService的init方法（子类NettyBlockTransferService实现了init方法），这个方法的作用就是初始化Netty服务，为拉取block数据提供服务。第二步是调用shuffleClient的init方法，shuffleClient这个引用有可能是BlockTransferService有可能是ExternalShuffleClient，取决于我们的配置文件是否配置了externalShuffleServiceEnabled未开启状态，其实无论是哪种，都是为了对外提供服务，能够使block数据再节点之间流动起来。
  3. BlockManagerMaster调用registerBlockManager方法，向BlockManagerMaster（其实BlockManagerMasterEndpoint）发送BlockManager的注册请求。
  4. BlockManagerMaster（其实BlockManagerMasterEndpoint）接受到BlockManager的注册请求后。会调用register方法，开始注册Executor上的BlockManager，注册完成以后将BlockManagerId返回给对应Executor上的BlockManager。
3. BlockManager源码
  1. 当我们的程序启动的时候，首先会创建SparkContext对象，在创建SparkContext对象的时候就会调用_env.blockManager.initialize(_applicationId)创建BlockManager对象，这个BlockManager就是Driver上的BlockManager，它负责管理集群中Executor上的BlockManager
  2. SparkContext里创建BlockManager代码片段
    复制代码//为Driver创建BlockManager _env.blockManager.initialize(_applicationId)
```
1
2
//为Driver创建BlockManager
_env.blockManager.initialize(_applicationId)
```
    - 1
    - 2
  3. 创建Executor的时候，Executor内部会调用_env.blockManager.initialize(conf.getAppId)方法创建BlockManager
    复制代码if (!isLocal) { env.metricsSystem.registerSource(executorSource) env.blockManager.initialize(conf.getAppId) }
```
1
2
3
4
if (!isLocal) {
    env.metricsSystem.registerSource(executorSource)
    env.blockManager.initialize(conf.getAppId)
  }
```
    - 1
    - 2
    - 3
    - 4
  4. BlockManager类里的initialize方法，该方法作用是创建BlockManager，并且向BlockManagerMaster进行注册
    复制代码def initialize(appId: String): Unit = { //初始化BlockTransferService，其实是它的子类NettyBlockTransferService是下了init方法， //该方法的作用就是初始化传输服务，通过传输服务可以从不同的节点上拉取Block数据 blockTransferService.init(this) shuffleClient.init(appId) //设置block的复制分片策略，由spark.storage.replication.policy指定 blockReplicationPolicy = { val priorityClass = conf.get( "spark.storage.replication.policy", classOf[RandomBlockReplicationPolicy].getName) val clazz = Utils.classForName(priorityClass) val ret = clazz.newInstance.asInstanceOf[BlockReplicationPolicy] logInfo(s"Using $priorityClass for block replication policy") ret } //根据给定参数为对对应的Executor封装一个BlockManagerId对象（块存储的唯一标识） //executorID：executor的Id，blockTransferService.hostName：传输Block数据的服务的主机名 //blockTransferService.port：传输Block数据的服务的主机名 val id = BlockManagerId(executorId, blockTransferService.hostName, blockTransferService.port, None) //调用BlockManagerMaster的registerBlockManager方法向Driver上的BlockManagerMaster注册 val idFromMaster = master.registerBlockManager( id, maxMemory, slaveEndpoint) //更新BlockManagerId blockManagerId = if (idFromMaster != null) idFromMaster else id //判断是否开了外部shuffle服务 shuffleServerId = if (externalShuffleServiceEnabled) { logInfo(s"external shuffle service port = $externalShuffleServicePort") BlockManagerId(executorId, blockTransferService.hostName, externalShuffleServicePort) } else { blockManagerId } // 如果开启了外部shuffle服务，并且该节点是Driver的话就调用registerWithExternalShuffleServer方法 //将BlockManager注册在本地 if (externalShuffleServiceEnabled && !blockManagerId.isDriver) { registerWithExternalShuffleServer() } logInfo(s"Initialized BlockManager: $blockManagerId") }
```
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def initialize(appId: String): Unit = {
    //初始化BlockTransferService，其实是它的子类NettyBlockTransferService是下了init方法，
    //该方法的作用就是初始化传输服务，通过传输服务可以从不同的节点上拉取Block数据
    blockTransferService.init(this)
    shuffleClient.init(appId)

    //设置block的复制分片策略，由spark.storage.replication.policy指定
    blockReplicationPolicy = {
      val priorityClass = conf.get(
        "spark.storage.replication.policy", classOf[RandomBlockReplicationPolicy].getName)
      val clazz = Utils.classForName(priorityClass)
      val ret = clazz.newInstance.asInstanceOf[BlockReplicationPolicy]
      logInfo(s"Using $priorityClass for block replication policy")
      ret
    }

    //根据给定参数为对对应的Executor封装一个BlockManagerId对象（块存储的唯一标识）
    //executorID：executor的Id，blockTransferService.hostName：传输Block数据的服务的主机名
    //blockTransferService.port：传输Block数据的服务的主机名
    val id = BlockManagerId(executorId, blockTransferService.hostName, blockTransferService.port, None)

    //调用BlockManagerMaster的registerBlockManager方法向Driver上的BlockManagerMaster注册
    val idFromMaster = master.registerBlockManager(
      id,
      maxMemory,
      slaveEndpoint)
    //更新BlockManagerId
    blockManagerId = if (idFromMaster != null) idFromMaster else id

    //判断是否开了外部shuffle服务
    shuffleServerId = if (externalShuffleServiceEnabled) {
      logInfo(s"external shuffle service port = $externalShuffleServicePort")
      BlockManagerId(executorId, blockTransferService.hostName, externalShuffleServicePort)
    } else {
      blockManagerId
    }

    // 如果开启了外部shuffle服务，并且该节点是Driver的话就调用registerWithExternalShuffleServer方法
    //将BlockManager注册在本地
    if (externalShuffleServiceEnabled && !blockManagerId.isDriver) {
      registerWithExternalShuffleServer()
    }

    logInfo(s"Initialized BlockManager: $blockManagerId")
  }
```
    - 1
    - 2
    - 3
    - 4
    - 5
    - 6
    - 7
    - 8
    - 9
    - 10
    - 11
    - 12
    - 13
    - 14
    - 15
    - 16
    - 17
    - 18
    - 19
    - 20
    - 21
    - 22
    - 23
    - 24
    - 25
    - 26
    - 27
    - 28
    - 29
    - 30
    - 31
    - 32
    - 33
    - 34
    - 35
    - 36
    - 37
    - 38
    - 39
    - 40
    - 41
    - 42
    - 43
    - 44
    - 45
  5. BlockManagerMaster类里的registerBlockManager方法，向Driver发送RegisterBlockManager消息进行注册
    复制代码def registerBlockManager(blockManagerId: BlockManagerId,maxMemSize: Long,slaveEndpoint: RpcEndpointRef): BlockManagerId = { logInfo(s"Registering BlockManager $blockManagerId") //向Driver发送注册BlockManager请求 //blockManagerId：块存储的唯一标识，里边封装了该BlockManager所在的executorId，提供Netty服务的主机名和端口 //maxMemSize最大的内存 val updatedId = driverEndpoint.askWithRetry[BlockManagerId]( RegisterBlockManager(blockManagerId, maxMemSize, slaveEndpoint)) logInfo(s"Registered BlockManager $updatedId") updatedId }
```
1
2
3
4
5
6
7
8
9
10
def registerBlockManager(blockManagerId: BlockManagerId,maxMemSize: Long,slaveEndpoint: RpcEndpointRef): BlockManagerId = {
    logInfo(s"Registering BlockManager $blockManagerId")
    //向Driver发送注册BlockManager请求
    //blockManagerId：块存储的唯一标识，里边封装了该BlockManager所在的executorId，提供Netty服务的主机名和端口
    //maxMemSize最大的内存
    val updatedId = driverEndpoint.askWithRetry[BlockManagerId](
      RegisterBlockManager(blockManagerId, maxMemSize, slaveEndpoint))
    logInfo(s"Registered BlockManager $updatedId")
    updatedId
  }
```
    - 1
    - 2
    - 3
    - 4
    - 5
    - 6
    - 7
    - 8
    - 9
    - 10
  6. BlockManagerMasterEndpoint类里的receiveAndReply方法，这个方法就是接受请求的消息，然后并处理请求
    复制代码def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = { //BlocManagerMasterEndPoint接收到来自Executor上的BlockManager注册请求的时候， //调用register方法开始注册BlockManager， case RegisterBlockManager(blockManagerId, maxMemSize, slaveEndpoint) => context.reply(register(blockManagerId, maxMemSize, slaveEndpoint)) //....其余代码省略 }
```
1
2
3
4
5
6
7
def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
    //BlocManagerMasterEndPoint接收到来自Executor上的BlockManager注册请求的时候，
    //调用register方法开始注册BlockManager，
    case RegisterBlockManager(blockManagerId, maxMemSize, slaveEndpoint) =>
      context.reply(register(blockManagerId, maxMemSize, slaveEndpoint))
    //....其余代码省略
  }
```
    - 1
    - 2
    - 3
    - 4
    - 5
    - 6
    - 7
  7. BlockManagerMasterEndpoint类里的register方法，该方法的作用就是开始注册executor上的BlockManager
    复制代码// BlockManager到BlockManaInfo的映射 private val blockManagerInfo = new mutable.HashMap[BlockManagerId, BlockManagerInfo] //executorId到BlockManagerId的映射 private val blockManagerIdByExecutor = new mutable.HashMap[String, BlockManagerId] private def register( idWithoutTopologyInfo: BlockManagerId, maxMemSize: Long, slaveEndpoint: RpcEndpointRef): BlockManagerId = { //利用从Executor上传过来的BlockManagerId信息重新封装BlockManagerId，并且 //之前传过来没有拓扑信息，这次直接将拓扑信息也封装进去，得到一个更完整的BlockManagerId val id = BlockManagerId( idWithoutTopologyInfo.executorId, idWithoutTopologyInfo.host, idWithoutTopologyInfo.port, topologyMapper.getTopologyForHost(idWithoutTopologyInfo.host)) val time = System.currentTimeMillis() //判断当前这个BlockManagerId是否注册过，注册结构为：HashMap[BlockManagerId, BlockManagerInfo] //如果没注册过就向下执行开始注册 if (!blockManagerInfo.contains(id)) { //首先会根据executorId查找内存缓存结构中是否有对应的BlockManagerId,如果为存在那么就将调用removeExecutor方法， //将executor从BlockManagerMaster中移除，首先会移除executorId对应的BlockManagerId,然后在移除该旧的BlockManager //其实就是移除以前的注册过的旧数据 blockManagerIdByExecutor.get(id.executorId) match { case Some(oldId) => // A block manager of the same executor already exists, so remove it (assumed dead) logError("Got two different block manager registrations on same executor - " + s" will replace old one $oldId with new one $id") removeExecutor(id.executorId) case None => } logInfo("Registering block manager %s with %s RAM, %s".format( id.hostPort, Utils.bytesToString(maxMemSize), id)) //将executorId与BlockManagerId映射起来，放到内存缓存中 blockManagerIdByExecutor(id.executorId) = id //将BlockManagerId与BlockManagerInfo映射起来，放入内存缓存中 //BlockManagerInfo封住了BlockMangerId，当前注册的事件，最大的内存 blockManagerInfo(id) = new BlockManagerInfo( id, System.currentTimeMillis(), maxMemSize, slaveEndpoint) } listenerBus.post(SparkListenerBlockManagerAdded(time, id, maxMemSize)) id }
```
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

// BlockManager到BlockManaInfo的映射
private val blockManagerInfo = new mutable.HashMap[BlockManagerId, BlockManagerInfo]

//executorId到BlockManagerId的映射
private val blockManagerIdByExecutor = new mutable.HashMap[String, BlockManagerId]

    private def register(
      idWithoutTopologyInfo: BlockManagerId,
      maxMemSize: Long,
      slaveEndpoint: RpcEndpointRef): BlockManagerId = {
    //利用从Executor上传过来的BlockManagerId信息重新封装BlockManagerId，并且
    //之前传过来没有拓扑信息，这次直接将拓扑信息也封装进去，得到一个更完整的BlockManagerId
    val id = BlockManagerId(
      idWithoutTopologyInfo.executorId,
      idWithoutTopologyInfo.host,
      idWithoutTopologyInfo.port,
      topologyMapper.getTopologyForHost(idWithoutTopologyInfo.host))

    val time = System.currentTimeMillis()
    //判断当前这个BlockManagerId是否注册过，注册结构为：HashMap[BlockManagerId, BlockManagerInfo]
    //如果没注册过就向下执行开始注册
    if (!blockManagerInfo.contains(id)) {
      //首先会根据executorId查找内存缓存结构中是否有对应的BlockManagerId,如果为存在那么就将调用removeExecutor方法，
      //将executor从BlockManagerMaster中移除，首先会移除executorId对应的BlockManagerId,然后在移除该旧的BlockManager
      //其实就是移除以前的注册过的旧数据
      blockManagerIdByExecutor.get(id.executorId) match {
        case Some(oldId) =>
          // A block manager of the same executor already exists, so remove it (assumed dead)
          logError("Got two different block manager registrations on same executor - "
              + s" will replace old one $oldId with new one $id")
          removeExecutor(id.executorId)
        case None =>
      }
      logInfo("Registering block manager %s with %s RAM, %s".format(
        id.hostPort, Utils.bytesToString(maxMemSize), id))

      //将executorId与BlockManagerId映射起来，放到内存缓存中
      blockManagerIdByExecutor(id.executorId) = id

      //将BlockManagerId与BlockManagerInfo映射起来，放入内存缓存中
      //BlockManagerInfo封住了BlockMangerId，当前注册的事件，最大的内存
      blockManagerInfo(id) = new BlockManagerInfo(
        id, System.currentTimeMillis(), maxMemSize, slaveEndpoint)
    }
    listenerBus.post(SparkListenerBlockManagerAdded(time, id, maxMemSize))
    id
  }
```
    - 1
    - 2
    - 3
    - 4
    - 5
    - 6
    - 7
    - 8
    - 9
    - 10
    - 11
    - 12
    - 13
    - 14
    - 15
    - 16
    - 17
    - 18
    - 19
    - 20
    - 21
    - 22
    - 23
    - 24
    - 25
    - 26
    - 27
    - 28
    - 29
    - 30
    - 31
    - 32
    - 33
    - 34
    - 35
    - 36
    - 37
    - 38
    - 39
    - 40
    - 41
    - 42
    - 43
    - 44
    - 45
    - 46
    - 47
    - 48
  8. BlockManagerMasterEndpoint类里的removeExecutor方法，该方法的作用就是移除掉之前注册过的旧数据
    复制代码
```
1
 
```
  9. 2. BlockManager定义
      1. BlockManager是Spark的分布式存储系统，与我们平常说的分布式存储系统是有区别的，区别就是这个分布式存储系统只会管理Block块数据，它运行在所有节点上。BlockManager的结构是Maser-Slave架构，Master就是Driver上的BlockManagerMaster，Slave就是每个Executor上的BlockManager。BlockManagerMaster负责接受Executor上的BlockManager的注册以及管理BlockManager的元数据信息
    3. BlockManager原理
      1. 从上边的定义我们已经得知，BlockManager是分布式的，运行在各个节点上的。从BlockManager的创建过程来看，其实Block是运行在Driver和每个Executor的。因为在创建SparkContext的时候，会调用SparkEnv.blockManager.initialize方法实例化BlockManager对象，在创建Executor对象的时候也会创建BlockManager。
      2. 在初始化BlockManager的时候，第一步会初始化BlockTransferService的init方法（子类NettyBlockTransferService实现了init方法），这个方法的作用就是初始化Netty服务，为拉取block数据提供服务。第二步是调用shuffleClient的init方法，shuffleClient这个引用有可能是BlockTransferService有可能是ExternalShuffleClient，取决于我们的配置文件是否配置了externalShuffleServiceEnabled未开启状态，其实无论是哪种，都是为了对外提供服务，能够使block数据再节点之间流动起来。
      3. BlockManagerMaster调用registerBlockManager方法，向BlockManagerMaster（其实BlockManagerMasterEndpoint）发送BlockManager的注册请求。
      4. BlockManagerMaster（其实BlockManagerMasterEndpoint）接受到BlockManager的注册请求后。会调用register方法，开始注册Executor上的BlockManager，注册完成以后将BlockManagerId返回给对应Executor上的BlockManager。
    4. BlockManager源码
      1. 当我们的程序启动的时候，首先会创建SparkContext对象，在创建SparkContext对象的时候就会调用_env.blockManager.initialize(_applicationId)创建BlockManager对象，这个BlockManager就是Driver上的BlockManager，它负责管理集群中Executor上的BlockManager
      2. SparkContext里创建BlockManager代码片段
        
        复制代码//为Driver创建BlockManager _env.blockManager.initialize(_applicationId)
        1
        2
        //为Driver创建BlockManager _env.blockManager.initialize(_applicationId)
        
        1
        2
      3. 创建Executor的时候，Executor内部会调用_env.blockManager.initialize(conf.getAppId)方法创建BlockManager
        
        复制代码if (!isLocal) { env.metricsSystem.registerSource(executorSource) env.blockManager.initialize(conf.getAppId) }
        1
        2
        3
        4
        if (!isLocal) { env.metricsSystem.registerSource(executorSource) env.blockManager.initialize(conf.getAppId) }
        
        1
        2
        3
        4
      4. BlockManager类里的initialize方法，该方法作用是创建BlockManager，并且向BlockManagerMaster进行注册
        
        复制代码def initialize(appId: String): Unit = { //初始化BlockTransferService，其实是它的子类NettyBlockTransferService是下了init方法， //该方法的作用就是初始化传输服务，通过传输服务可以从不同的节点上拉取Block数据 blockTransferService.init(this) shuffleClient.init(appId) //设置block的复制分片策略，由spark.storage.replication.policy指定 blockReplicationPolicy = { val priorityClass = conf.get( "spark.storage.replication.policy", classOf[RandomBlockReplicationPolicy].getName) val clazz = Utils.classForName(priorityClass) val ret = clazz.newInstance.asInstanceOf[BlockReplicationPolicy] logInfo(s"Using $priorityClass for block replication policy") ret } //根据给定参数为对对应的Executor封装一个BlockManagerId对象（块存储的唯一标识） //executorID：executor的Id，blockTransferService.hostName：传输Block数据的服务的主机名 //blockTransferService.port：传输Block数据的服务的主机名 val id = BlockManagerId(executorId, blockTransferService.hostName, blockTransferService.port, None) //调用BlockManagerMaster的registerBlockManager方法向Driver上的BlockManagerMaster注册 val idFromMaster = master.registerBlockManager( id, maxMemory, slaveEndpoint) //更新BlockManagerId blockManagerId = if (idFromMaster != null) idFromMaster else id //判断是否开了外部shuffle服务 shuffleServerId = if (externalShuffleServiceEnabled) { logInfo(s"external shuffle service port = $externalShuffleServicePort") BlockManagerId(executorId, blockTransferService.hostName, externalShuffleServicePort) } else { blockManagerId } // 如果开启了外部shuffle服务，并且该节点是Driver的话就调用registerWithExternalShuffleServer方法 //将BlockManager注册在本地 if (externalShuffleServiceEnabled && !blockManagerId.isDriver) { registerWithExternalShuffleServer() } logInfo(s"Initialized BlockManager: $blockManagerId") }
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23
        24
        25
        26
        27
        28
        29
        30
        31
        32
        33
        34
        35
        36
        37
        38
        39
        40
        41
        42
        43
        44
        45
        def initialize(appId: String): Unit = { //初始化BlockTransferService，其实是它的子类NettyBlockTransferService是下了init方法， //该方法的作用就是初始化传输服务，通过传输服务可以从不同的节点上拉取Block数据 blockTransferService.init(this) shuffleClient.init(appId) //设置block的复制分片策略，由spark.storage.replication.policy指定 blockReplicationPolicy = { val priorityClass = conf.get( "spark.storage.replication.policy", classOf[RandomBlockReplicationPolicy].getName) val clazz = Utils.classForName(priorityClass) val ret = clazz.newInstance.asInstanceOf[BlockReplicationPolicy] logInfo(s"Using $priorityClass for block replication policy") ret } //根据给定参数为对对应的Executor封装一个BlockManagerId对象（块存储的唯一标识） //executorID：executor的Id，blockTransferService.hostName：传输Block数据的服务的主机名 //blockTransferService.port：传输Block数据的服务的主机名 val id = BlockManagerId(executorId, blockTransferService.hostName, blockTransferService.port, None) //调用BlockManagerMaster的registerBlockManager方法向Driver上的BlockManagerMaster注册 val idFromMaster = master.registerBlockManager( id, maxMemory, slaveEndpoint) //更新BlockManagerId blockManagerId = if (idFromMaster != null) idFromMaster else id //判断是否开了外部shuffle服务 shuffleServerId = if (externalShuffleServiceEnabled) { logInfo(s"external shuffle service port = $externalShuffleServicePort") BlockManagerId(executorId, blockTransferService.hostName, externalShuffleServicePort) } else { blockManagerId } // 如果开启了外部shuffle服务，并且该节点是Driver的话就调用registerWithExternalShuffleServer方法 //将BlockManager注册在本地 if (externalShuffleServiceEnabled && !blockManagerId.isDriver) { registerWithExternalShuffleServer() } logInfo(s"Initialized BlockManager: $blockManagerId") }
        
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23
        24
        25
        26
        27
        28
        29
        30
        31
        32
        33
        34
        35
        36
        37
        38
        39
        40
        41
        42
        43
        44
        45
      5. BlockManagerMaster类里的registerBlockManager方法，向Driver发送RegisterBlockManager消息进行注册
        
        复制代码def registerBlockManager(blockManagerId: BlockManagerId,maxMemSize: Long,slaveEndpoint: RpcEndpointRef): BlockManagerId = { logInfo(s"Registering BlockManager $blockManagerId") //向Driver发送注册BlockManager请求 //blockManagerId：块存储的唯一标识，里边封装了该BlockManager所在的executorId，提供Netty服务的主机名和端口 //maxMemSize最大的内存 val updatedId = driverEndpoint.askWithRetry[BlockManagerId]( RegisterBlockManager(blockManagerId, maxMemSize, slaveEndpoint)) logInfo(s"Registered BlockManager $updatedId") updatedId }
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        def registerBlockManager(blockManagerId: BlockManagerId,maxMemSize: Long,slaveEndpoint: RpcEndpointRef): BlockManagerId = { logInfo(s"Registering BlockManager $blockManagerId") //向Driver发送注册BlockManager请求 //blockManagerId：块存储的唯一标识，里边封装了该BlockManager所在的executorId，提供Netty服务的主机名和端口 //maxMemSize最大的内存 val updatedId = driverEndpoint.askWithRetry[BlockManagerId]( RegisterBlockManager(blockManagerId, maxMemSize, slaveEndpoint)) logInfo(s"Registered BlockManager $updatedId") updatedId }
        
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
      6. BlockManagerMasterEndpoint类里的receiveAndReply方法，这个方法就是接受请求的消息，然后并处理请求
        
        复制代码def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = { //BlocManagerMasterEndPoint接收到来自Executor上的BlockManager注册请求的时候， //调用register方法开始注册BlockManager， case RegisterBlockManager(blockManagerId, maxMemSize, slaveEndpoint) => context.reply(register(blockManagerId, maxMemSize, slaveEndpoint)) //....其余代码省略 }
        1
        2
        3
        4
        5
        6
        7
        def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = { //BlocManagerMasterEndPoint接收到来自Executor上的BlockManager注册请求的时候， //调用register方法开始注册BlockManager， case RegisterBlockManager(blockManagerId, maxMemSize, slaveEndpoint) => context.reply(register(blockManagerId, maxMemSize, slaveEndpoint)) //....其余代码省略 }
        
        1
        2
        3
        4
        5
        6
        7
      7. BlockManagerMasterEndpoint类里的register方法，该方法的作用就是开始注册executor上的BlockManager
        
        复制代码// BlockManager到BlockManaInfo的映射 private val blockManagerInfo = new mutable.HashMap[BlockManagerId, BlockManagerInfo] //executorId到BlockManagerId的映射 private val blockManagerIdByExecutor = new mutable.HashMap[String, BlockManagerId] private def register( idWithoutTopologyInfo: BlockManagerId, maxMemSize: Long, slaveEndpoint: RpcEndpointRef): BlockManagerId = { //利用从Executor上传过来的BlockManagerId信息重新封装BlockManagerId，并且 //之前传过来没有拓扑信息，这次直接将拓扑信息也封装进去，得到一个更完整的BlockManagerId val id = BlockManagerId( idWithoutTopologyInfo.executorId, idWithoutTopologyInfo.host, idWithoutTopologyInfo.port, topologyMapper.getTopologyForHost(idWithoutTopologyInfo.host)) val time = System.currentTimeMillis() //判断当前这个BlockManagerId是否注册过，注册结构为：HashMap[BlockManagerId, BlockManagerInfo] //如果没注册过就向下执行开始注册 if (!blockManagerInfo.contains(id)) { //首先会根据executorId查找内存缓存结构中是否有对应的BlockManagerId,如果为存在那么就将调用removeExecutor方法， //将executor从BlockManagerMaster中移除，首先会移除executorId对应的BlockManagerId,然后在移除该旧的BlockManager //其实就是移除以前的注册过的旧数据 blockManagerIdByExecutor.get(id.executorId) match { case Some(oldId) => // A block manager of the same executor already exists, so remove it (assumed dead) logError("Got two different block manager registrations on same executor - " + s" will replace old one $oldId with new one $id") removeExecutor(id.executorId) case None => } logInfo("Registering block manager %s with %s RAM, %s".format( id.hostPort, Utils.bytesToString(maxMemSize), id)) //将executorId与BlockManagerId映射起来，放到内存缓存中 blockManagerIdByExecutor(id.executorId) = id //将BlockManagerId与BlockManagerInfo映射起来，放入内存缓存中 //BlockManagerInfo封住了BlockMangerId，当前注册的事件，最大的内存 blockManagerInfo(id) = new BlockManagerInfo( id, System.currentTimeMillis(), maxMemSize, slaveEndpoint) } listenerBus.post(SparkListenerBlockManagerAdded(time, id, maxMemSize)) id }
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23
        24
        25
        26
        27
        28
        29
        30
        31
        32
        33
        34
        35
        36
        37
        38
        39
        40
        41
        42
        43
        44
        45
        46
        47
        48
        // BlockManager到BlockManaInfo的映射 private val blockManagerInfo = new mutable.HashMap[BlockManagerId, BlockManagerInfo] //executorId到BlockManagerId的映射 private val blockManagerIdByExecutor = new mutable.HashMap[String, BlockManagerId] private def register( idWithoutTopologyInfo: BlockManagerId, maxMemSize: Long, slaveEndpoint: RpcEndpointRef): BlockManagerId = { //利用从Executor上传过来的BlockManagerId信息重新封装BlockManagerId，并且 //之前传过来没有拓扑信息，这次直接将拓扑信息也封装进去，得到一个更完整的BlockManagerId val id = BlockManagerId( idWithoutTopologyInfo.executorId, idWithoutTopologyInfo.host, idWithoutTopologyInfo.port, topologyMapper.getTopologyForHost(idWithoutTopologyInfo.host)) val time = System.currentTimeMillis() //判断当前这个BlockManagerId是否注册过，注册结构为：HashMap[BlockManagerId, BlockManagerInfo] //如果没注册过就向下执行开始注册 if (!blockManagerInfo.contains(id)) { //首先会根据executorId查找内存缓存结构中是否有对应的BlockManagerId,如果为存在那么就将调用removeExecutor方法， //将executor从BlockManagerMaster中移除，首先会移除executorId对应的BlockManagerId,然后在移除该旧的BlockManager //其实就是移除以前的注册过的旧数据 blockManagerIdByExecutor.get(id.executorId) match { case Some(oldId) => // A block manager of the same executor already exists, so remove it (assumed dead) logError("Got two different block manager registrations on same executor - " + s" will replace old one $oldId with new one $id") removeExecutor(id.executorId) case None => } logInfo("Registering block manager %s with %s RAM, %s".format( id.hostPort, Utils.bytesToString(maxMemSize), id)) //将executorId与BlockManagerId映射起来，放到内存缓存中 blockManagerIdByExecutor(id.executorId) = id //将BlockManagerId与BlockManagerInfo映射起来，放入内存缓存中 //BlockManagerInfo封住了BlockMangerId，当前注册的事件，最大的内存 blockManagerInfo(id) = new BlockManagerInfo( id, System.currentTimeMillis(), maxMemSize, slaveEndpoint) } listenerBus.post(SparkListenerBlockManagerAdded(time, id, maxMemSize)) id }
        
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23
        24
        25
        26
        27
        28
        29
        30
        31
        32
        33
        34
        35
        36
        37
        38
        39
        40
        41
        42
        43
        44
        45
        46
        47
        48
      8. BlockManagerMasterEndpoint类里的removeExecutor方法，该方法的作用就是移除掉之前注册过的旧数据
        
        复制代码
        1