MapReduce:计算Job运行时的CPU与内存平均利用率

    Hadoop集群上运行有多道MapReduce Job，到底每个Job消费集群计算资源(CPU，内存)的情况如何，通常我们是不知道的。这节来探讨下如果想获得这些信息，应该怎样做。

    每个Job在运行时是散布在TaskTracker上，由不同的JVM来执行。所以这个问题归根结底就是想获取每个运行task的JVM对资源的消费情况。对于这种获取分布式环境数据的例子，就可以采用MapReduce框架自带的Counter机制，由它来采集各JVM运行期的实时数据，然后在Job结束后，做最终分析。

这项任务的意义在哪里？
    正常情况下，Job会因为数据范围、服务的应用、数据量的大小及MapReduce逻辑的复杂程度等因素而被定义为不同的类别。当类别的概念形成后，我们就想知道每个类别Job对集群资源的使用情况是否符合预期，如果可以获得这些数据，就可以根据资源使用情况对Job的执行时间做相应的调整，以避免对其它Job的影响。

数据采集什么时候开始与结束？
    我们的目标是分析每个子JVM的资源消耗情况，就应该在子JVM开始运行时介入收集。站在MapReduce框架外，只有一个地方可以作为介入点：Mapper和Reducer的setup方法。它的参数是task运行期Context，可以很方便地调用Counter。那我们的入口就可以像这样

public class ResourceCollectMapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
							extends Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {

	ChildJVMResourceCollector collector = null;
	public void setup(Context context) {
		collector = new ChildJVMResourceCollector(context);
		collector.start();
	}
	
}

    资源收集器是一个独立线程，在setup方法中启动，在cleanup方法调用时结束。本来整个task的执行过程就包含在setup与cleanup之间，所以我们完全有机会收集这个过程中的资源利用情况。

如何获取运行task的JVM pid？
    运行task的子JVM是由TaskTracker创建的，对用户完全透明。但幸运的是，TaskTrack在创建这个子JVM时，把它的pid作为参数传给了子JVM。获取的方式如下

private String getChildJVMPid() {
	String pid = "";
	if (!Shell.WINDOWS) {
		pid = System.getenv().get("JVM_PID");
	}
	return pid;
}

怎样根据pid获取系统级的资源使用率？
    得到当前子JVM的进程号后，就能用这个进程号查询Linux系统当前进程的资源使用情况。查询的方式网上很多资料都有说明



    上图是通过ps命令查询某个进程的CPU与内存使用情况的截图。其它相似的查询方式大家可以自行实验
    如果采用上述的linux命令，就需要分开组装，并调用Hadoop提供的Shell组件。

String[] queryCommand = new String[]{"ps", "-o", "%cpu,%mem", pid};
String result = Shell.execCommand(queryCommand);

返回的查询结果是一个字符串，解析字符串，拿到想要的结果，以特定的key把结果存入Counter中(当然获取次数也需要收集，在Counter中由独立key标识)，这样我们的一次数据获取过程就结束了。

最终怎样分析？
    在Job结束后，根据特定的key获取相应的数据，然后除以获取次数，就是平均的CPU与内存使用率。

要注意的情况是？
    在这个过程中需要注意有两个地方：
    1. 尽量让获取资源的线程不要占用太多的执行时间，每隔一段时间启用一次，减少对正常task执行的影响；
    2. 如果遇到某个Job的多个task共用一个子JVM的情况(MapReduce优化的一点，减少频繁创建/销毁JVM的代价)，就需要确保在当前task的cleanup方法中一定让获取资源的线程终结掉。这个线程包含着task运行期的Context，如果是Map task的话，那个默认100MB的内存缓冲区也在其中，可能会有OOM问题而影响下一个task的创建过程。

评论

5 楼 farseeing 2013-06-10  

楼主你好，既然在TaskTracker中可以获得Child Jvm的pid，为什么不直接在TaskTracker中起一个线程来监测Child Jvm的资源利用率呢？

4 楼 langyu 2012-10-22  

xilinniao 写道

数据采集点的设置是在：Mapper和Reducer的setup方法。难道在setup（）方法中就已经决定了该task（mapper或reduce）所需要的资源，而不是在map（）或reduce（）方法中决定task的资源使用情况？？

只是在setup中启动一个独立线程，用来收集task运行（map/reduce执行过程）中的资源使用情况。状态收集与map/reduce的执行是并发运行

3 楼 xilinniao 2012-10-22  

数据采集点的设置是在：Mapper和Reducer的setup方法。难道在setup（）方法中就已经决定了该task（mapper或reduce）所需要的资源，而不是在map（）或reduce（）方法中决定task的资源使用情况？？

2 楼 langyu 2012-04-16  

scu_cxh 写道

上面说的在map/reduce类中的setup方法开始收集，然后通过JVM_ID获取占用的资源信息，能否说的具体点呀？。。

你好，不知道你想收集哪些信息。我的作法是在Child JVM执行mapper的setup方法时，初始化一个新的线程，这个线程用来收集Child JVM执行期间的各种信息。然后信息通过Counter汇总。在Job结束后分析Counter结果，得到你想要的。如有不明白请联系[e-mail]

1 楼 scu_cxh 2012-04-13  

博主，您好，我最近也在看Hadoop方面的东西，想做一个去监控task运行情况的东西，但是没有什么思路，看了你的文章之后，有点启发，不过还是有点模糊，上面说的在map/reduce类中的setup方法开始收集，然后通过JVM_ID获取占用的资源信息，能否说的具体点呀？。。