Hadoop2源碼分析－YARN RPC 示例介紹

時間  2019-11-18
標籤 hadoop2 hadoop 源碼 分析 yarn rpc 示例 介紹 欄目 Hadoop 简体版
原文   原文鏈接

1.概述

　　以前在《Hadoop2源碼分析－RPC探索實戰》一文當中介紹了Hadoop的RPC機制，今天給你們分享關於YARN的RPC的機制。下面是今天的分享目錄：html

  • YARN的RPC介紹
  • YARN的RPC示例
  • 截圖預覽

　　下面開始今天的內容分享。java

2.YARN的RPC介紹

　　咱們知道在Hadoop的RPC當中，其主要由RPC，Client及Server這三個大類組成，分別實現對外提供編程接口、客戶端實現及服務端實現。以下圖所示：node

 

 

　　圖中是Hadoop的RPC的一個類的關係圖，你們能夠到《Hadoop2源碼分析－RPC探索實戰》一文中，經過代碼示例去理解他們之間的關係，這裏就很少作贅述了。接下來，咱們去看Yarn的RPC。apache

　　Yarn對外提供的是YarnRPC這個類，這是一個抽象類，經過閱讀YarnRPC的源碼能夠知道，實際的實現由參數yarn.ipc.rpc.class設定，默認狀況下，其值爲：org.apache.hadoop.yarn.ipc.HadoopYarnProtoRPC，部分代碼以下：編程

  • YarnRPC：
public abstract class YarnRPC {
   // ......

    public static YarnRPC create(Configuration conf) {
    LOG.debug("Creating YarnRPC for " + 
        conf.get(YarnConfiguration.IPC_RPC_IMPL));
    String clazzName = conf.get(YarnConfiguration.IPC_RPC_IMPL);
    if (clazzName == null) {
      clazzName = YarnConfiguration.DEFAULT_IPC_RPC_IMPL;
    }
    try {
      return (YarnRPC) Class.forName(clazzName).newInstance();
    } catch (Exception e) {
      throw new YarnRuntimeException(e);
    }
  }

}
  • YarnConfiguration類：
public class YarnConfiguration extends Configuration {

  //Configurations
  public static final String YARN_PREFIX = "yarn.";

  ////////////////////////////////
  // IPC Configs
  ////////////////////////////////
  public static final String IPC_PREFIX = YARN_PREFIX + "ipc.";
  /** RPC class implementation*/
  public static final String IPC_RPC_IMPL =
    IPC_PREFIX + "rpc.class";
  public static final String DEFAULT_IPC_RPC_IMPL = 
    "org.apache.hadoop.yarn.ipc.HadoopYarnProtoRPC";
}

　　而HadoopYarnProtoRPC 經過 RPC 的 RpcFactoryProvider 生成客戶端工廠(由參數 yarn.ipc.client.factory.class 指定，默認值是 org.apache.hadoop.yarn.factories.impl.pb.RpcClientFactoryPBImpl)和服務器工廠 (由參數 yarn.ipc.server.factory.class 指定，默認值是 org.apache.hadoop.yarn.factories.impl.pb.RpcServerFactoryPBImpl)，以根據通訊協議的 Protocol Buffers 定義生成客戶端對象和服務器對象。相關類的部分代碼以下：服務器

  • HadoopYarnProtoRPC
public class HadoopYarnProtoRPC extends YarnRPC {

  private static final Log LOG = LogFactory.getLog(HadoopYarnProtoRPC.class);

  @Override
  public Object getProxy(Class protocol, InetSocketAddress addr,
      Configuration conf) {
    LOG.debug("Creating a HadoopYarnProtoRpc proxy for protocol " + protocol);
    return RpcFactoryProvider.getClientFactory(conf).getClient(protocol, 1,
        addr, conf);
  }

  @Override
  public void stopProxy(Object proxy, Configuration conf) {
    RpcFactoryProvider.getClientFactory(conf).stopClient(proxy);
  }

  @Override
  public Server getServer(Class protocol, Object instance,
      InetSocketAddress addr, Configuration conf,
      SecretManager<? extends TokenIdentifier> secretManager,
      int numHandlers, String portRangeConfig) {
    LOG.debug("Creating a HadoopYarnProtoRpc server for protocol " + protocol + 
        " with " + numHandlers + " handlers");
    
    return RpcFactoryProvider.getServerFactory(conf).getServer(protocol, 
        instance, addr, conf, secretManager, numHandlers, portRangeConfig);

  }

}

  • RpcFactoryProvideride

public class RpcFactoryProvider {

  // ......

  public static RpcClientFactory getClientFactory(Configuration conf) {
    String clientFactoryClassName = conf.get(
        YarnConfiguration.IPC_CLIENT_FACTORY_CLASS,
        YarnConfiguration.DEFAULT_IPC_CLIENT_FACTORY_CLASS);
    return (RpcClientFactory) getFactoryClassInstance(clientFactoryClassName);
  }

  //......
  
}
/** Factory to create client IPC classes.*/
  public static final String IPC_CLIENT_FACTORY_CLASS =
    IPC_PREFIX + "client.factory.class";
  public static final String DEFAULT_IPC_CLIENT_FACTORY_CLASS = 
      "org.apache.hadoop.yarn.factories.impl.pb.RpcClientFactoryPBImpl";

　　在 YARN 中並未使用Hadoop自帶的Writable來作序列化，而是使用 Protocol Buffers 做爲默認的序列化機制，這帶來的好處主要有如下幾點：函數

  • 繼承Protocol Buffers的優勢：Protocol Buffers已被實踐證實其擁有高效性、可擴展性、緊湊性以及跨語言性等特色。
  • 支持在線升級回滾：在Hadoop 2.x版本後，添加的HA方案，該方案可以進行主備切換，在不中止NNA節點服務的前提下，可以在線升級版本。

3.YARN的RPC示例

　　YARN 的工做流程是先定義通訊協議接口ResourceTracker，它包含2個函數，具體代碼以下所示：oop

  • ResourceTracker：
public interface ResourceTracker {
  
  @Idempotent
  public RegisterNodeManagerResponse registerNodeManager(
      RegisterNodeManagerRequest request) throws YarnException,
      IOException;

  @AtMostOnce
  public NodeHeartbeatResponse nodeHeartbeat(NodeHeartbeatRequest request)
      throws YarnException, IOException;

}

　　這裏ResourceTracker提供了Protocol Buffers定義和Java實現，其中設計的Protocol Buffers文件有：ResourceTracker.proto、yarn_server_common_service_protos.proto和yarn_server_common_protos.proto，文件路徑在Hadoop的源碼包的 hadoop-2.6.0-src/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto，這裏就不貼出3個文件的具體代碼類，你們能夠到該目錄去閱讀這部分代碼。這裏須要注意的是，如果你們要編譯這些文件須要安裝 ProtoBuf 的編譯環境，環境安裝較爲簡單，這裏給你們簡要說明下。源碼分析

　　首先是下載ProtoBuf的安裝包，而後解壓，進入到解壓目錄，編譯安裝。命令以下：

./configure --prefix=/home/work /protobuf/  

make && make install

最後編譯 .proto 文件的命令：

protoc ./ResourceTracker.proto  --java_out=./

　　下面，咱們去收取Hadoop源碼到本地工程，運行調試相關代碼。

  • TestYarnServerApiClasses：

public class TestYarnServerApiClasses {

  // ......

  // 列舉測試4個方法  

@Test
  public void testRegisterNodeManagerResponsePBImpl() {
    RegisterNodeManagerResponsePBImpl original =
        new RegisterNodeManagerResponsePBImpl();
    original.setContainerTokenMasterKey(getMasterKey());
    original.setNMTokenMasterKey(getMasterKey());
    original.setNodeAction(NodeAction.NORMAL);
    original.setDiagnosticsMessage("testDiagnosticMessage");

    RegisterNodeManagerResponsePBImpl copy =
        new RegisterNodeManagerResponsePBImpl(
            original.getProto());
    assertEquals(1, copy.getContainerTokenMasterKey().getKeyId());
    assertEquals(1, copy.getNMTokenMasterKey().getKeyId());
    assertEquals(NodeAction.NORMAL, copy.getNodeAction());
    assertEquals("testDiagnosticMessage", copy.getDiagnosticsMessage());

  }

@Test
  public void testNodeHeartbeatRequestPBImpl() {
    NodeHeartbeatRequestPBImpl original = new NodeHeartbeatRequestPBImpl();
    original.setLastKnownContainerTokenMasterKey(getMasterKey());
    original.setLastKnownNMTokenMasterKey(getMasterKey());
    original.setNodeStatus(getNodeStatus());
    NodeHeartbeatRequestPBImpl copy = new NodeHeartbeatRequestPBImpl(
        original.getProto());
    assertEquals(1, copy.getLastKnownContainerTokenMasterKey().getKeyId());
    assertEquals(1, copy.getLastKnownNMTokenMasterKey().getKeyId());
    assertEquals("localhost", copy.getNodeStatus().getNodeId().getHost());
  }

@Test
  public void testNodeHeartbeatResponsePBImpl() {
    NodeHeartbeatResponsePBImpl original = new NodeHeartbeatResponsePBImpl();

    original.setDiagnosticsMessage("testDiagnosticMessage");
    original.setContainerTokenMasterKey(getMasterKey());
    original.setNMTokenMasterKey(getMasterKey());
    original.setNextHeartBeatInterval(1000);
    original.setNodeAction(NodeAction.NORMAL);
    original.setResponseId(100);

    NodeHeartbeatResponsePBImpl copy = new NodeHeartbeatResponsePBImpl(
        original.getProto());
    assertEquals(100, copy.getResponseId());
    assertEquals(NodeAction.NORMAL, copy.getNodeAction());
    assertEquals(1000, copy.getNextHeartBeatInterval());
    assertEquals(1, copy.getContainerTokenMasterKey().getKeyId());
    assertEquals(1, copy.getNMTokenMasterKey().getKeyId());
    assertEquals("testDiagnosticMessage", copy.getDiagnosticsMessage());
  }

@Test
  public void testRegisterNodeManagerRequestPBImpl() {
    RegisterNodeManagerRequestPBImpl original = new RegisterNodeManagerRequestPBImpl();
    original.setHttpPort(8080);
    original.setNodeId(getNodeId());
    Resource resource = recordFactory.newRecordInstance(Resource.class);
    resource.setMemory(10000);
    resource.setVirtualCores(2);
    original.setResource(resource);
    RegisterNodeManagerRequestPBImpl copy = new RegisterNodeManagerRequestPBImpl(
        original.getProto());

    assertEquals(8080, copy.getHttpPort());
    assertEquals(9090, copy.getNodeId().getPort());
    assertEquals(10000, copy.getResource().getMemory());
    assertEquals(2, copy.getResource().getVirtualCores());

  }

}

  • TestResourceTrackerPBClientImpl：

public class TestResourceTrackerPBClientImpl {

    private static ResourceTracker client;
    private static Server server;
    private final static org.apache.hadoop.yarn.factories.RecordFactory recordFactory = RecordFactoryProvider
            .getRecordFactory(null);

    @BeforeClass
    public static void start() {

        System.out.println("Start client test");

        InetSocketAddress address = new InetSocketAddress(0);
        Configuration configuration = new Configuration();
        ResourceTracker instance = new ResourceTrackerTestImpl();
        server = RpcServerFactoryPBImpl.get().getServer(ResourceTracker.class, instance, address, configuration, null,
                1);
        server.start();

        client = (ResourceTracker) RpcClientFactoryPBImpl.get().getClient(ResourceTracker.class, 1,
                NetUtils.getConnectAddress(server), configuration);

    }

    @AfterClass
    public static void stop() {

        System.out.println("Stop client");

        if (server != null) {
            server.stop();
        }
    }

    /**
     * Test the method registerNodeManager. Method should return a not null
     * result.
     * 
     */
    @Test
    public void testResourceTrackerPBClientImpl() throws Exception {
        RegisterNodeManagerRequest request = recordFactory.newRecordInstance(RegisterNodeManagerRequest.class);
        assertNotNull(client.registerNodeManager(request));

        ResourceTrackerTestImpl.exception = true;
        try {
            client.registerNodeManager(request);
            fail("there should be YarnException");
        } catch (YarnException e) {
            assertTrue(e.getMessage().startsWith("testMessage"));
        } finally {
            ResourceTrackerTestImpl.exception = false;
        }

    }

    /**
     * Test the method nodeHeartbeat. Method should return a not null result.
     * 
     */

    @Test
    public void testNodeHeartbeat() throws Exception {
        NodeHeartbeatRequest request = recordFactory.newRecordInstance(NodeHeartbeatRequest.class);
        assertNotNull(client.nodeHeartbeat(request));

        ResourceTrackerTestImpl.exception = true;
        try {
            client.nodeHeartbeat(request);
            fail("there  should be YarnException");
        } catch (YarnException e) {
            assertTrue(e.getMessage().startsWith("testMessage"));
        } finally {
            ResourceTrackerTestImpl.exception = false;
        }

    }

    public static class ResourceTrackerTestImpl implements ResourceTracker {

        public static boolean exception = false;

        public RegisterNodeManagerResponse registerNodeManager(RegisterNodeManagerRequest request)
                throws YarnException, IOException {
            if (exception) {
                throw new YarnException("testMessage");
            }
            return recordFactory.newRecordInstance(RegisterNodeManagerResponse.class);
        }

        public NodeHeartbeatResponse nodeHeartbeat(NodeHeartbeatRequest request) throws YarnException, IOException {
            if (exception) {
                throw new YarnException("testMessage");
            }
            return recordFactory.newRecordInstance(NodeHeartbeatResponse.class);
        }

    }
}

4.截圖預覽

　　接下來，咱們使用JUnit去測試代碼，截圖預覽以下所示：

  • 對testRegisterNodeManagerRequestPBImpl()方法的一個DEBUG調試

  • testResourceTrackerPBClientImpl()方法的DEBUG調試

　　這裏因爲設置exception的狀態爲true，在調用registerNodeManager()時，會打印一條測試異常信息。

if (exception) {
　　throw new YarnException("testMessage");
}

5.總結

　　在學習Hadoop YARN的RPC時，能夠先了解Hadoop的RPC機制，這樣在接觸YARN的RPC的會比較好理解，YARN的RPC只是其中的一部分，後續會給你們分享更多關於YARN的內容。

6.結束語

　　這篇博客就和你們分享到這裏，若是你們在研究學習的過程中有什麼問題，能夠加羣進行討論或發送郵件給我，我會盡我所能爲您解答，與君共勉！

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程