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集群进行通信，完成诸如作业提交、作业状态查询等过程，通过它获取集群的信息。