MapReduce源码分析之新API作业提交（二）：连接集群

MapReduce作业提交时连接集群是通过Job的connect()方法实现的，它实际上是构造集群Cluster实例cluster，代码如下：

 private synchronized void connect()
         throws IOException, InterruptedException, ClassNotFoundException {

// 如果cluster为null，构造Cluster实例cluster，
// Cluster为连接MapReduce集群的一种工具，提供了一种获取MapReduce集群信息的方法
if (cluster == null) {
     cluster =
       ugi.doAs(new PrivilegedExceptionAction<Cluster>() {
                  public Cluster run()
                         throws IOException, InterruptedException,
                                ClassNotFoundException {
                    return new Cluster(getConfiguration());
                  }
                });
   }
 }

        这个方法用synchronized关键字标识，处理逻辑为：如果cluster为null，构造Cluster实例cluster。

        Cluster为连接MapReduce集群的一种工具，提供了一种获取MapReduce集群信息的方法，我们看下它的成员变量，如下所示：

// 客户端通信协议提供者
private ClientProtocolProvider clientProtocolProvider;
// 客户端通信协议实例
private ClientProtocol client;

// 用户信息
private UserGroupInformation ugi;

// 配置信息
private Configuration conf;

// 文件系统实例
private FileSystem fs = null;

// 系统路径
private Path sysDir = null;

// 阶段区域路径
private Path stagingAreaDir = null;

// 作业历史路径
private Path jobHistoryDir = null;

// 日志
private static final Log LOG = LogFactory.getLog(Cluster.class);

// 客户端通信协议提供者加载器
private static ServiceLoader<ClientProtocolProvider> frameworkLoader =
    ServiceLoader.load(ClientProtocolProvider.class);

        Cluster最重要的两个成员变量是客户端通信协议提供者ClientProtocolProvider实例clientProtocolProvider，客户端通信协议ClientProtocol实例client，而后者是依托前者的create()方法生成的。

        Cluster提供了两个构造函数，如下：

  public Cluster(Configuration conf) throws IOException {
    this(null, conf);
  }

  public Cluster(InetSocketAddress jobTrackAddr, Configuration conf)
      throws IOException {
<span style="white-space:pre">    </span>// 设置配置信息
    this.conf = conf;

    // 获取当前用户
    this.ugi = UserGroupInformation.getCurrentUser();

    // 调用initialize()方法完成初始化
    initialize(jobTrackAddr, conf);
  }

        最终会调用initialize()方法完成初始化，代码如下：

// 确定客户端ClientProtocol实例client
private void initialize(InetSocketAddress jobTrackAddr, Configuration conf)
    throws IOException {

  synchronized (frameworkLoader) {

    // 取出每个ClientProtocolProvider实例provider，通过其create()方法，
    // 构造ClientProtocol实例clientProtocol，
    // 并将两者赋值给对类应成员变量，退出循环
    for (ClientProtocolProvider provider : frameworkLoader) {

    LOG.debug("Trying ClientProtocolProvider : "
          + provider.getClass().getName());

    ClientProtocol clientProtocol = null;

    try {

        // 通过ClientProtocolProvider的create()方法，获取客户端与集群通讯ClientProtocol实例clientProtocol
        if (jobTrackAddr == null) {
          clientProtocol = provider.create(conf);
        } else {
          clientProtocol = provider.create(jobTrackAddr, conf);
        }

        // 设置类成员变量clientProtocolProvider、client，并退出循环
        if (clientProtocol != null) {
          clientProtocolProvider = provider;
          client = clientProtocol;

          // 记录debug级别日志信息
          LOG.debug("Picked " + provider.getClass().getName()
              + " as the ClientProtocolProvider");
          break;
        }
        else {

        // 记录debug级别日志信息
          LOG.debug("Cannot pick " + provider.getClass().getName()
              + " as the ClientProtocolProvider - returned null protocol");
        }
      }
      catch (Exception e) {
        LOG.info("Failed to use " + provider.getClass().getName()
            + " due to error: " + e.getMessage());
      }
    }
  }

  // 如果clientProtocolProvider、client任一为空，直接抛出IO异常
  if (null == clientProtocolProvider || null == client) {
    throw new IOException(
        "Cannot initialize Cluster. Please check your configuration for "
            + MRConfig.FRAMEWORK_NAME
            + " and the correspond server addresses.");
  }
}

        initialize()方法唯一的一个任务就是确定客户端通信协议提供者clientProtocolProvider，并通过其create()方法构造客户端通信协议ClientProtocol实例client。

        MapReduce中，ClientProtocolProvider抽象类的实现共有YarnClientProtocolProvider、LocalClientProtocolProvider两种，前者为Yarn模式，而后者为Local模式。

        我们先看下Yarn模式，看下YarnClientProtocolProvider的create()方法，代码如下：

 @Override
 public ClientProtocol create(Configuration conf) throws IOException {

// 如果参数mapreduce.framework.name配置的为yarn，构造一个YARNRunner实例并返回，否则返回null
   if (MRConfig.YARN_FRAMEWORK_NAME.equals(conf.get(MRConfig.FRAMEWORK_NAME))) {
     return new YARNRunner(conf);
   }
   return null;
 }

        Yarn模式下，如果参数mapreduce.framework.name配置的为yarn，构造一个YARNRunner实例并返回，否则返回null，关于YARNRunner，我们待会再讲，我们接着再看下Local模式，LocalClientProtocolProvider的create()方法，代码如下：

 @Override
 public ClientProtocol create(Configuration conf) throws IOException {

// 初始化framework：取参数mapreduce.framework.name，参数未配置默认为local
String framework =
       conf.get(MRConfig.FRAMEWORK_NAME, MRConfig.LOCAL_FRAMEWORK_NAME);

// 如果framework是local，，则返回LocalJobRunner实例，并设置map任务数量为1，否则返回null
   if (!MRConfig.LOCAL_FRAMEWORK_NAME.equals(framework)) {
     return null;
   }
   conf.setInt(JobContext.NUM_MAPS, 1);

   return new LocalJobRunner(conf);
 }

        Local模式也是需要看参数mapreduce.framework.name的配置是否为local，是的话，返回LocalJobRunner实例，并设置map任务数量为1，否则返回null，值得一提的是，这里参数mapreduce.framework.name未配置的话，默认为local，也就是说，MapReduce需要看参数mapreduce.framework.name确定连接模式，但默认是Local模式的。

        到了这里，我们就能够知道一个很重要的信息，Cluster中客户端通信协议ClientProtocol实例，要么是Yarn模式下的YARNRunner，要么就是Local模式下的LocalJobRunner，记住这点，对透彻了解MapReduce作业提交的整体流程非常重要。

        好了，我们继续以Yarn模式来分析MapReduce集群连接，看下YARNRunner的实现，先看下它的成员变量，如下：

// 记录工厂RecordFactory实例
private final RecordFactory recordFactory = RecordFactoryProvider.getRecordFactory(null);

// ResourceManager代理ResourceMgrDelegate实例
private ResourceMgrDelegate resMgrDelegate;

// 客户端缓存ClientCache实例
private ClientCache clientCache;

// 配置信息Configuration实例
private Configuration conf;

// 文件上下文FileContext实例
private final FileContext defaultFileContext;

        其中，最重要的一个变量就是ResourceManager代理ResourceMgrDelegate实例resMgrDelegate，Yarn模式下整个MapReduce客户端就是由它负责与Yarn集群进行通信，完成诸如作业提交、作业状态查询等过程，通过它获取集群的信息，其内部有一个YarnClient实例YarnClient，负责与Yarn进行通信，还有ApplicationId、ApplicationSubmissionContext等与特定应用程序相关的成员变量。关于ResourceMgrDelegate的详细介绍，请阅读《MapReduce源码分析ResourceMgrDelegate》一文，这里不再做详细介绍。

        另外一个比较重要的变量就是客户端缓存ClientCache实例clientCache，

        接下来，我们看下YARNRunner的构造函数，如下：

 /**
  * Yarn runner incapsulates the client interface of
  * yarn
  * @param conf the configuration object for the client
  */
 public YARNRunner(Configuration conf) {

  // 先构造ResourceManager代理ResourceMgrDelegate实例，然后再调用两个参数的构造函数
  this(conf, new ResourceMgrDelegate(new YarnConfiguration(conf)));
 }

 /**
  * Similar to {@link #YARNRunner(Configuration)} but allowing injecting
  * {@link ResourceMgrDelegate}. Enables mocking and testing.
  * @param conf the configuration object for the client
  * @param resMgrDelegate the resourcemanager client handle.
  */
 public YARNRunner(Configuration conf, ResourceMgrDelegate resMgrDelegate) {
  // 先构造客户端缓存ClientCache实例，然后再调用三个参数的构造函数
  this(conf, resMgrDelegate, new ClientCache(conf, resMgrDelegate));
 }

 /**
  * Similar to {@link YARNRunner#YARNRunner(Configuration, ResourceMgrDelegate)}
  * but allowing injecting {@link ClientCache}. Enable mocking and testing.
  * @param conf the configuration object
  * @param resMgrDelegate the resource manager delegate
  * @param clientCache the client cache object.
  */
 public YARNRunner(Configuration conf, ResourceMgrDelegate resMgrDelegate,
     ClientCache clientCache) {

// 成员变量赋值
this.conf = conf;
   try {
     this.resMgrDelegate = resMgrDelegate;
     this.clientCache = clientCache;

     // 获取文件山下文FileContext实例defaultFileContext
     this.defaultFileContext = FileContext.getFileContext(this.conf);
   } catch (UnsupportedFileSystemException ufe) {
     throw new RuntimeException("Error in instantiating YarnClient", ufe);
   }
 }

        YARNRunner一共提供了三个构造函数，而我们之前说的WordCount作业提交时，其内部调用的是YARNRunner带有一个参数的构造函数，它会先构造ResourceManager代理ResourceMgrDelegate实例，然后再调用两个参数的构造函数，继而构造客户端缓存ClientCache实例，然后再调用三个参数的构造函数，而最终的构造函数只是进行简单的类成员变量赋值，然后通过FileContext的静态getFileContext()方法获取文件山下文FileContext实例defaultFileContext。

        总结

        MapReduce作业提交时连接集群是通过Job的connect()方法实现的，它实际上是构造集群Cluster实例cluster。Cluster为连接MapReduce集群的一种工具，提供了一种获取MapReduce集群信息的方法。在Cluster内部，有一个与集群进行通信的客户端通信协议ClientProtocol实例client，它由ClientProtocolProvider的静态create()方法构造，而Hadoop2.6.0中提供了两种模式的ClientProtocol，分别为Yarn模式的YARNRunner和Local模式的LocalJobRunner，Cluster实际上是由它们负责与集群进行通信的，而Yarn模式下，ClientProtocol实例YARNRunner对象内部有一个ResourceManager代理ResourceMgrDelegate实例resMgrDelegate，Yarn模式下整个MapReduce客户端就是由它负责与Yarn集群进行通信，完成诸如作业提交、作业状态查询等过程，通过它获取集群的信息。