Hadoop全链路监控解决方案

前言

我在最近的几篇文章中都或多或少的提到了一个很重要的词-"监控".为什么要提到这个词呢,因为如果你和我一样是一名大数据工程师,你手下管理着批量的集群机器,并且同时这个集群的规模还会不定时的扩大,机器一旦变多,发生问题的频率和类型就会变多,所以这是,你靠人肉去分析某台机器上的日志,OK,1台机器,2台机器,尚且可以解决办法,但是100台,1000台呢,当然如果工程师还这么做的话,我想他会抓狂的.所以如何做到智能化发现问题,定位问题,就显得很关键了,最理想的结果是,你拥有你的集群机器中每天跑的job的各种指标数据,然后你动动鼠标,通过展示出来的图形界面,就迅速的发现了问题.这也是正是我最近在做的一件事情,效果还算不错.下面是我近1个月来,对于我们部门的Hadoop集群做的一些监控方面的事情,谈不上高大上的结构,我们是如何以最简单的方式达到最大化的效果,希望能给大家带来帮助.

监控两大主题

监控的大方向是想好了,但是我们要监控哪些指标,有必要提一下,要想做好监控,首先你要了解Hadoop这一整套的系统逻辑,大致了解即可,那么我们是怎么做的呢.分为2大方向,宏观层面和微观层面.宏观角度的理解就是Node级别,拓扑结构级别,DataNode,NameNode,JournalNode,ResourceManager,NodeManager,主要就是这5大组件,通过分析这些节点上的监控数据,一般你能够定位到慢节点,可能某台机器的网络出问题了,或者说某台机器执行的时间总是大于正常机器等等这样类似的问题.刚刚说的是一个监控方向,还有1个是微观层面,就是细粒度化的监控,基于user用户级别,基于单个job,单个task级别的监控,像这类监控指标就是另一大方向,这类的监控指标在实际的使用场景中特别重要,一旦你的集群资源是开放给外面的用户使用,用户本身不了解你的这套机制原理,很容易会乱申请资源,造成严重拖垮集群整体运作效率的事情,所以这类监控的指标就是为了防止这样的事情发生.

宏观层面监控

OK,两大主题监控方向确定,下面谈谈具体的实践方案,想必这也是大家所关心的.首先要先想能不能通过最简单的方式直接拿到一些数据,比如RPC接口去取,这样的方式的确是有的,我们用了YarnClient的获取nodeReport方法,做了这么一个获取工具,下面是核心代码:
YarnClient client;
client = YarnClient.createYarnClient();
client.init(new Configuration());
client.start();

List<NodeReport> nodesReport = client.getNodeReports();
...这样就得到nodeManager中的许多数据,包括正在运行的container数量,使用的内存大小,核数多少等等信息,同理DataNode的数据也可以以类似的方式获取,这里就不给出代码了.在分节点的统计代码中,还有一项指标也可以帮助我们发现节点异常问题,我们对各个IP对namenode的rpc请求量做了统计,对namenode的请求其实就是对HDFS的请求最了统计.当然,这个统计肯定要自己去分析才能得到.分析日志中的哪条记录信息呢,如果你没有阅读过NameNode和FSNameSystem这2个类的代码,你可能会不知道,hadoop在设计的时候,考虑的比较好,对于Client的每次请求,都做了请求记录的输出,名字叫做audit日志:
/**
* Default AuditLogger implementation; used when no access logger is
* defined in the config file. It can also be explicitly listed in the
* config file.
*/
private static class DefaultAuditLogger extends HdfsAuditLogger {
....

@Override
public void logAuditEvent(boolean succeeded, String userName,
InetAddress addr, String cmd, String src, String dst,
FileStatus status, UserGroupInformation ugi,
DelegationTokenSecretManager dtSecretManager) {

if (auditLog.isInfoEnabled()) {
final StringBuilder sb = auditBuffer.get();
sb.setLength(0);
sb.append("allowed=").append(succeeded).append("\t");
sb.append("ugi=").append(userName).append("\t");
sb.append("ip=").append(addr).append("\t");
sb.append("cmd=").append(cmd).append("\t");
sb.append("src=").append(src).append("\t");
sb.append("dst=").append(dst).append("\t");
if (null == status) {
sb.append("perm=null");
} else {
sb.append("perm=");
sb.append(status.getOwner()).append(":");
sb.append(status.getGroup()).append(":");
sb.append(status.getPermission());
}
if (logTokenTrackingId) {
sb.append("\t").append("trackingId=");
String trackingId = null;
if (ugi != null && dtSecretManager != null
&& ugi.getAuthenticationMethod() == AuthenticationMethod.TOKEN) {
for (TokenIdentifier tid: ugi.getTokenIdentifiers()) {
if (tid instanceof DelegationTokenIdentifier) {
DelegationTokenIdentifier dtid =
(DelegationTokenIdentifier)tid;
trackingId = dtSecretManager.getTokenTrackingId(dtid);
break;
}
}
}
sb.append(trackingId);
}
sb.append("\t").append("proto=");
sb.append(NamenodeWebHdfsMethods.isWebHdfsInvocation() ? "webhdfs" : "rpc");
logAuditMessage(sb.toString());
}
}
里面的信息非常全,有用户者,操作命令,操作的源文件,目标文件,请求ip,其实单纯这几个信息,就可以帮助我们分析出很多有价值的信息了,当然我们先用他来分析基于IP的请求统计.这些记录如果你不做特殊的处理,他是零散的分布于nameode日志中的,所以你要把他从茫茫的记录中筛选出来,单独放到一个文件中,这个hadoop也是可以做到的,配置log4j的配置文件,增加下面的配置信息.
#
# hdfs audit logging
#
hdfs.audit.logger=INFO,RFAAUDIT
hdfs.audit.log.maxfilesize=256MB
hdfs.audit.log.maxbackupindex=100
log4j.logger.org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit=${hdfs.audit.logger}
log4j.additivity.org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit=false
log4j.appender.RFAAUDIT=org.apache.log4j.RollingFileAppender
log4j.appender.RFAAUDIT.File=${hadoop.log.dir}/hdfs-audit.log
log4j.appender.RFAAUDIT.layout=org.apache.log4j.PatternLayout
log4j.appender.RFAAUDIT.layout.ConversionPattern=%d{ISO8601} %p %c{2}: %m%n
log4j.appender.RFAAUDIT.MaxFileSize=${hdfs.audit.log.maxfilesize}
log4j.appender.RFAAUDIT.MaxBackupIndex=${hdfs.audit.log.maxbackupindex}这样就会单独的输出名字为hdfs-audit.log的文件数据了.分析的方式就是写个程序做文本解析,进行count计数即可.我们对这个文件还做了但分钟级别qps请求量的趋势统计,分用户级别的请求量分析,大家也可以这么去做.
总的来说,离线的统计方式基本可以采用这样的方式去做,配lo4j单独打出某个类的日志,单类文件进行解析.

微观层面监控

微观层面的监控就要比宏观层面的监控要复杂许多,首先同样你要了解一定的理论只是,在Hadoop中,一个job是以怎么样的方式执行,如何转化,到最后的任务结束.当然在这里不可能一次性讲清楚.现在只要知道,这么几个名次,Application,Job,Task,Container,我们所说的细粒度监控指的也就是对这些变量进行监控.当然这些指标的量级是很大的,Application应用数和Job数也是可以1天过万的,这些job生成的task小的十几,多则上千个task,总的也可能达到1天上百万个task在跑.对于微观层面的监控，采用分析日志的方式，就不见得会很适合，应该并没有一个完整的地方会把每次task的启动信息大全，所以我们需要借助一下Hadoop自身对job的监控，就是JobHistory,这个类中就保存了历史执行过的job的执行信息，1个job里面会有所有执行过得task信息 ，task里又会包含有很多task attempt的信息。我们可以模拟JobHistory解析jHistoryFile格式的job信息文件进行信息的分析。当然你可以自己找到job的Path,做纯文本的解读。这方面的内容可以阅读我的这篇文章，JobHistory的job信息获取与分析。在task级别还有1个可以做的方面是增加counter，增加更多的自定义counte,counter的指标可以更加多元化，比如task Gc次数，full Gc总次数，reduce过程最后产生的文件数等等，这些指标都能细粒度的反映出这个task的执行好坏，如何加自定义的counter,大家也可以阅读我的这篇文章自定义Counter添加。还有1个微观层次的监控是基于用户的监控，将获取到的task，job这些按照用户进行聚合，排序，进行二次分析，就能找到问题用户，使用资源的大户。

基于Hadoop源码的改造监控升级

这个监控改造是一个升级，补充了前者监控中的不足，因为前面2个都是属于离线分析，事后分析，我们当然需要实时的结果数据，当然我们不会再做一个类似的实时数据收集展示平台，我们直接改JobHistory界面，ResourceManager后台显示界面，展示更多的页面信息在界面上，老实说，目前的Hadoop版本中的一些Ganglia监控指标和JobHistoty的显示信息，实在有限，分为2种情况，Ganglia上是我们想要的监控目标上面没有，所以我们只能自己再加，JobHistory上则是指标太多，导致上面只显示了公共常见的一些信息，很多都是要点击到页面里才能看，因此我们做的改进是把里面的信息放出来。当然这些改造都是我们内部自己对Hadoop的源码进行改造的。

不足之处

1.这个不足之处是目前对于我们部门来说可以补充去做的地方，冷热数据的分析，这个也可以通过hdfs-audit日志文件，根据访问命中次数，然后动态调整相应的副本数，提高读写效率。
2.集群所有文件的结构分析，比如所有文件的大小范围分布情况，是不是小文件在集群中占了很大的比例，这个需要对fsimage文件做分析，这个我们之前也做过，但是没有做出工具，只是临时的解析了一下。
3.Container数据的获得，Container的数据在JobHistory上没有的，上面只有1个containerId。container的信息也很有用，上面有许多的优先级，还有附加的信息都可以了解到机器资源的使用情况等等。

工具分享链接

下面是我近端时间做分析工具的一些部分核心代码，供大家学习使用：
JobHistory文件手动分析工具：https://github.com/linyiqun/yarn-jobhistory-crawler/tree/master/jobHiveSqlAnalyse
NodeManager定时状态信息获取工具：https://github.com/linyiqun/yarn-jobhistory-crawler/tree/master/NMTool
JobHistory文件程序分析工具：https://github.com/linyiqun/yarn-jobhistory-crawler
HDFS RPC分用户，IP请求分析工具：https://github.com/linyiqun/yarn-auditlog-parser