hadoop集群升级到yarn遇到的数据本地化(data-locality)很低的问题分析

最近将hadoop集群从hadoop-2.0.0-mr1-cdh4.1.2升级到hadoop-2.0.0-cdh4.3.0,遇到了一些任务(如scan hbase table)数据本地化data-local很低的问题,原因就是前篇文章中说到的ipv4和ipv6导致的机器名长短问题。本文是对整个过程的分析。

我们的yarn集群采用的是capacity-scheduler调度,data-local很低的原因是:机器名不一致,导致分配container时无法匹配上。在yarn内部(NodeManager)用的机器名是 test042097.sqa.cm4 这种短的,而job的split信息(如scan hbase)里面带的机器名是 test042097.sqa.cm4.site.net 这种完整机器名。

在 ResourceManager 分配container的地方:(RMContainerAllocator.java:967行)

   private ContainerRequest assignToMap(Container allocated) {
   //try to assign to maps if present first by host,then by rack,followed by *
     ContainerRequest assigned = null;
     while (assigned == null && maps.size() > 0) {
       String host = allocated.getNodeId().getHost();
       LinkedList<TaskAttemptId> list = mapsHostMapping.get(host);
       while (list != null && list.size() > 0) {
         ……

         LOG.debug("Assigned based on host match " + host); // 1.主机名匹配

       }

       if (assigned == null) {
         String rack = RackResolver.resolve(host).getNetworkLocation();
         list = mapsRackMapping.get(rack);
         ……

          LOG.debug("Assigned based on rack match " + rack); // 2.机架匹配

         }
         if (assigned == null && maps.size() > 0) {
           TaskAttemptId tId = maps.keySet().iterator().next();
           ……
           if (LOG.isDebugEnabled()) {
             LOG.debug("Assigned based on * match");  // 3.任意匹配
           }
           break;
         }
    ……
}

代码里面是按照 host, rack, any 三种方式顺序匹配的,在 mapsHostMapping 里面存的是资源的请求,<hostname, List<TaskAttemptId>>,意思是对请求在某台机器上启动几个task。hostname是来自于job输入的split信息,比如扫描hbase表的job,split信息里面就有region所在的机器信息。

在hbase里面,用到的机器名都是 test042097.sqa.cm4.site.net 这种带了 site.net 后缀的,所以mapsHostMapping的key就是 test042097.sqa.cm4.site.net。而Yarn的NodeManager所用的机器名都是 test042097.sqa.cm4 这种不带后缀的,也就是 allocated.getNodeId().getHost();的值。因此机器名是没法匹配上的。


后来,我们的集群采用斯盛同学提出的改法:“每台机器一个rack”后,rack local比例提高了很多,此时的rack local其实就是data locale。

这样改后,上面的代码中 RackResolver.resolve(host).getNetworkLocation(); 这个会调用集群配置的rack脚本,会读取一个rack文件信息:

10.1.42.93 /rack/1
test042093.sqa.cm4 /rack/1
……
10.1.42.97 /rack/4
test042097.sqa.cm4 /rack/4

所以,通过search042097.sqa.cm4可以映射到 /rack/4,与 mapsRackMapping 中的数据就匹配上了。(mapsRackMapping中存有请求的资源rack信息,yarn中的资源请求是超额请求的,用一个申请三个,主机+机架+any)

YARN中的NodeManager获取hostname的地方,是在ContainerManagerImpl.java:243行

InetSocketAddress connectAddress = NetUtils.getConnectAddress(server);
this.context.getNodeId().setHost(connectAddress.getHostName());
LOG.info("ContainerManager started at " + connectAddress);

底层调用的是jdk的

InetAddress.getLocalHost().getHostName()

方法,得到的是 search042097.sqa.cm4 这种短主机名。

而普通map reduce job split信息中的机器名和HBase中RegionServer使用的机器名(zk上还有hbase master页面可以看到)都是长机器名 search042097.sqa.cm4.site.net。

hbase master的机器名获取是通过ip反查DNS得到的,参考代码如下:

String hostname = Strings.domainNamePointerToHostName(DNS.getDefaultHost("default", "default"));
InetSocketAddress initialIsa = new InetSocketAddress(hostname, 4000);
System.out.println("HBASE MASTER=" + initialIsa.getHostName() + "\n");

而通过 clientChannel.socket().getInetAddress().getHostName() 获取到的也是完整的机器名(写了个java nio的程序模拟mater-regionserver进行了验证)。region server启动log里面也会有显示:

hbase-hadoop-regionserver-search042024.sqa.cm4.log.2013-08-07:2013-08-07 13:06:21,200 INFO org.apache.hadoop.hbase.regionserver.HRegionServer: Master passed us hostname to use. Was=search042024.sqa.cm4, Now=search042024.sqa.cm4.site.net

为什么升级前的hadoop版本没有问题呢?

问了下公司的JVM大牛坤谷,java中“InetAddress.getLocalHost().getHostName()” 获取机器名有没有什么特别之处。坤谷提到有这个一个“bug”:JDK-7166687 : InetAddress.getLocalHost().getHostName() returns FQDN。说的是在不同jdk上获取的机器名返回不一样,里面讲到“Run with IPv6 disabled, -Djava.net.preferIPv4Stack=true”会影响机器名的输出。

查了下hadoop的启动参数,发现对于老版本的hadoop,hadoop-env.sh里面有这样一行:

export HADOOP_OPTS="-Djava.net.preferIPv4Stack=true $HADOOP_CLIENT_OPTS

而最新的yarn启动脚本里面并没有加上这个参数。

于是做了些试验:在开启了ipv6的机器上,由于会默认采用ipv6,通过 InetAddress.getLocalHost().getHostName() 获取到的是短机器名,加上-Djava.net.preferIPv4Stack=true 运行就能获取到长机器名了。而这个调用在只有ipv4的机器上获取的都是长机器名。后来也关闭了某些机器的ipv6来进行验证。

查了下线上yarn集群的机器,发现部分新的机器上开启了ipv6,老的机器并没有开启ipv6,这也解释了在yarn的/cluster/nodes页面上看到的机器名是有的长,有的短。

因此对于机器ipv4和ipv6不统一的集群来说,在 yarn-env.sh 里面加上 YARN_OPTS=”$YARN_OPTS -Djava.net.preferIPv4Stack=true” 就能使得YARN里面获取的机器名都是长机器名,与hbase中的一致,从而使得按照主机名能匹配上,提高data-local比例。


后续:往YARN社区提交了个issue:Inconsistent hostname leads to low data locality on IPv6 hosts。看来Hadoop YARN确实没在ipv6上做过充分测试,官方也不推荐在hadoop集群上使用ipv6: http://wiki.apache.org/hadoop/HadoopIPv6

但是我觉得,在大规模分布式的hadoop集群中,机器都是异构的,很大几率会出现新老机器ipv4和ipv6配置不一致的情况,所以在YARN里面强制指定ipv4或者用ipv6确实是有必要的。


发现在hadoop-2.2.0-cdh5.0.0-beta版本里面已经改掉了,采用了和IPV4,IPV6无关的getCanonicalHostName调用:

 InetSocketAddress connectAddress = NetUtils.getConnectAddress(server);
    NodeId nodeId = NodeId.newInstance(
        connectAddress.getAddress().getCanonicalHostName(),
        connectAddress.getPort());
    ((NodeManager.NMContext)context).setNodeId(nodeId);
    this.context.getNMTokenSecretManager().setNodeId(nodeId);
    this.context.getContainerTokenSecretManager().setNodeId(nodeId);
    LOG.info("ContainerManager started at " + connectAddress);

发表评论

电子邮件地址不会被公开。 必填项已用*标注