Hadoop自定义JobTracker和NameNode管理页面

为了可以方便地在hadoop的管理界面（namenode和jobtracker）中自定义展示项，使用代理servlet的方式实现了hadoop的管理界面。

首先，

在org.apache.hadoop.http.HttpServer中的构造函数public HttpServer(String name, String bindAddress, int port,boolean findPort, Configuration conf, AccessControlList adminsAcl,Connector connector)中添加如下代码，指定资源包和url前缀：

ServletHolder servletHoder = new ServletHolder();

servletHoder.setInitParameter("com.sun.jersey.config.property.packages","com.test.agent");

servletHoder.setServlet(new com.sun.jersey.spi.container.servlet.ServletContainer());

webAppContext.addServlet(servletHoder, "/agent/*");

之后，在包中定义了一个抽象对象AbstractResource　

public class AbstractResource {

	protected JsonMapper jsonMapper = JsonMapper.nonDefaultMapper();

	protected @Context

	ServletContext context;

	public JsonMapper getJsonMapper() {

		return jsonMapper;

	}

	public void setJsonMapper(JsonMapper jsonMapper) {

		this.jsonMapper = jsonMapper;

	}

	public ServletContext getContext() {

		return context;

	}

	public void setContext(ServletContext context) {

		this.context = context;

	}

}

然后，真实的资源代码实现AbstractResource对象，对资源进行定义

@Path("/job")

public class JobResource extends AbstractResource {

	private DecimalFormat percentFormat = new DecimalFormat("##0.00");

	private SimpleDateFormat dateFormat = new SimpleDateFormat(

			"d-MMM-yyyy HH:mm:ss");

	@Path("/clustersummary")

	@GET

	@Consumes(MediaType.TEXT_PLAIN)

	@Produces(MediaType.TEXT_PLAIN)

	public String getClusterSummaryInfo() {

		JobTracker jt = (JobTracker) context.getAttribute("job.tracker");

		ClusterMetrics metrics = jt.getClusterMetrics();

		String tasksPerNode = metrics.getTaskTrackerCount() > 0 ? percentFormat

				.format(((double) (metrics.getMapSlotCapacity() + metrics

						.getReduceSlotCapacity()))

						/ metrics.getTaskTrackerCount()) : "-";

		Map<String, String> summaryInfo = new LinkedHashMap<String, String>();

		summaryInfo.put("usedHeapMemoryBytes",

				Long.toString(Runtime.getRuntime().totalMemory()));

		summaryInfo.put("totalHeapMemoryBytes",

				Long.toString(Runtime.getRuntime().maxMemory()));

		summaryInfo.put("runningMapTasks",

				Integer.toString(metrics.getRunningMaps()));

		summaryInfo.put("runningReduceTasks",

				Integer.toString(metrics.getRunningReduces()));

		summaryInfo.put("totalJobSubmissions",

				Integer.toString(metrics.getTotalJobSubmissions()));

		summaryInfo.put("numTotalTaskTrackers",

				Integer.toString(metrics.getTaskTrackerCount()));

		summaryInfo.put("occupiedMapSlots",

				Integer.toString(metrics.getOccupiedMapSlots()));

		summaryInfo.put("occupiedReduceSlots",

				Integer.toString(metrics.getOccupiedReduceSlots()));

		summaryInfo.put("reservedMapSlots",

				Integer.toString(metrics.getReservedMapSlots()));

		summaryInfo.put("reservedReduceSlots",

				Integer.toString(metrics.getReservedReduceSlots()));

		summaryInfo.put("mapTaskCapacity",

				Integer.toString(metrics.getMapSlotCapacity()));

		summaryInfo.put("reduceTaskCapacity",

				Integer.toString(metrics.getReduceSlotCapacity()));

		summaryInfo.put("avgTasksPerTaskTracker", tasksPerNode);

		summaryInfo.put("numBlackListedTaskTrackers",

				Integer.toString(metrics.getBlackListedTaskTrackerCount()));

		summaryInfo.put("numGrayListedTaskTrackers",

				Integer.toString(metrics.getGrayListedTaskTrackerCount()));

		summaryInfo.put("numDecommissionedTaskTrackers",

				Integer.toString(metrics.getDecommissionedTaskTrackerCount()));

		return jsonMapper.toJson(summaryInfo);

	}

}

这样可以通过访问url进行访问，自定义管理界面访问的端口与原有的管理页面端口相同。JobTracker使用50030，NameNode使用50070，依次类推。具体见下单元测试代码：

public class JobResourceIntTest {

	HttpClient client;

	@Before

	public void setUp() {

		client = new DefaultHttpClient();

	}

	private String urlPrefix = "http://ip:50030/agent/job/";

	private String jobId = "job_201405121106_0001";

	private String tipId = "task_201405121106_0001_m_000000";

	private String attemptId = "attempt_201405121106_0001_m_000000_0";

	private String regex = "\"total\":\"(.+?)\",";

	private Pattern p = Pattern.compile(regex);

	@Test

	public void testClusterSummary() {

		HttpGet clusterSummary = new HttpGet(urlPrefix + "clustersummary");

		try {

			HttpResponse response = client.execute(clusterSummary);

			String clusterSummaryStr = EntityUtils.toString(response

					.getEntity());

			System.out.println(clusterSummaryStr);

			Assert.assertTrue(clusterSummaryStr.contains("usedHeapMemoryBytes"));

			Assert.assertTrue(clusterSummaryStr.contains("runningMapTasks"));

		} catch (Exception e) {

			e.printStackTrace();

		}

		clusterSummary.releaseConnection();

	}

	@After

	public void tearDown() {

	}

}

上述是后端的实现和单元测试，前端可根据需要进行资源获取和界面设计。　

巴特西

Hadoop自定义JobTracker和NameNode管理页面

最新文章

热门文章