文档章节

Storm【实践系列-如何写一个爬虫- Metric 系列】1

止静
 止静
发布于 2014/08/21 16:19
字数 620
阅读 803
收藏 0
package com.digitalpebble.storm.crawler;

import backtype.storm.Config;
import backtype.storm.metric.MetricsConsumerBolt;
import backtype.storm.metric.api.IMetricsConsumer;
import backtype.storm.task.IErrorReporter;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;
import backtype.storm.utils.Utils;
import com.google.common.base.Joiner;
import com.google.common.base.Supplier;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSortedMap;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.ObjectWriter;
import org.mortbay.jetty.Server;
import org.mortbay.jetty.servlet.Context;
import org.mortbay.jetty.servlet.ServletHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Enno Shioji (enno.shioji@peerindex.com)
 */
public class DebugMetricConsumer implements IMetricsConsumer {
	private static final Logger log = LoggerFactory
			.getLogger(DebugMetricConsumer.class);
	private IErrorReporter errorReporter;
	private Server server;

	// Make visible to servlet threads
	private volatile TopologyContext context;
	private volatile ConcurrentMap<String, Number> metrics;
	private volatile ConcurrentMap<String, Map<String, Object>> metrics_metadata;

	public void prepare(Map stormConf, Object registrationArgument,
			TopologyContext context, IErrorReporter errorReporter) {
		this.context = context;
		this.errorReporter = errorReporter;
		this.metrics = new ConcurrentHashMap<String, Number>();
		this.metrics_metadata = new ConcurrentHashMap<String, Map<String, Object>>();

		try {
			// TODO Config file not tested
			final String PORT_CONFIG_STRING = "topology.metrics.consumers.debug.servlet.port";
			Integer port = (Integer) stormConf.get(PORT_CONFIG_STRING);
			if (port == null) {
				log.warn("Metrics debug servlet's port not specified, defaulting to 7070. You can specify it via "
						+ PORT_CONFIG_STRING + " in storm.yaml");
				port = 7070;
			}
			server = startServlet(port);
		} catch (Exception e) {
			log.error("Failed to start metrics server", e);
			throw new AssertionError(e);
		}
	}

	private static final Joiner ON_COLONS = Joiner.on("::");

	public void handleDataPoints(TaskInfo taskInfo,
			Collection<DataPoint> dataPoints) {
		// In order
		String componentId = taskInfo.srcComponentId;
		Integer taskId = taskInfo.srcTaskId;
		Integer updateInterval = taskInfo.updateIntervalSecs;
		Long timestamp = taskInfo.timestamp;
		for (DataPoint point : dataPoints) {
			String metric_name = point.name;
			try {
				Map<String, Number> metric = (Map<String, Number>) point.value;
				for (Map.Entry<String, Number> entry : metric.entrySet()) {
					String metricId = ON_COLONS.join(componentId, taskId,
							metric_name, entry.getKey());
					Number val = entry.getValue();
					metrics.put(metricId, val);
					metrics_metadata.put(metricId, ImmutableMap
							.<String, Object> of("updateInterval",
									updateInterval, "lastreported", timestamp));
				}
			} catch (RuntimeException e) {
				// One can easily send something else than a Map<String,Number>
				// down the __metrics stream and make this part break.
				// If you ask me either the message should carry type
				// information or there should be different stream per message
				// type
				// This is one of the reasons why I want to write a further
				// abstraction on this facility
				errorReporter.reportError(e);
				metrics_metadata
						.putIfAbsent("ERROR_METRIC_CONSUMER_"
								+ e.getClass().getSimpleName(), ImmutableMap
								.of("offending_message_sample", point.value));
			}
		}
	}

	private static final ObjectMapper OM = new ObjectMapper();

	private Server startServlet(int serverPort) throws Exception {
		// Setup HTTP server
		Server server = new Server(serverPort);
		Context root = new Context(server, "/");
		server.start();

		HttpServlet servlet = new HttpServlet() {
			@Override
			protected void doGet(HttpServletRequest req,
					HttpServletResponse resp) throws ServletException,
					IOException {
				SortedMap<String, Number> metrics = ImmutableSortedMap
						.copyOf(DebugMetricConsumer.this.metrics);
				SortedMap<String, Map<String, Object>> metrics_metadata = ImmutableSortedMap
						.copyOf(DebugMetricConsumer.this.metrics_metadata);

				Map<String, Object> toplevel = ImmutableMap
						.of("retrieved",
								new Date(),

								// TODO this call fails with mysterious
								// exception
								// "java.lang.IllegalArgumentException: Could not find component common for __metrics"
								// Mailing list suggests it's a library version
								// issue but couldn't find anything suspicious
								// Need to eventually investigate
								// "sources",
								// context.getThisSources().toString(),

								"metrics", metrics, "metric_metadata",
								metrics_metadata);

				ObjectWriter prettyPrinter = OM
						.writerWithDefaultPrettyPrinter();
				prettyPrinter.writeValue(resp.getWriter(), toplevel);
			}
		};

		root.addServlet(new ServletHolder(servlet), "/metrics");

		log.info("Started metric server...");
		return server;

	}

	public void cleanup() {
		try {
			server.stop();
		} catch (Exception e) {
			throw new AssertionError(e);
		}
	}

}

 


   前提说明:

          storm从0.9.0开始,增加了指标统计框架,用来收集应用程序的特定指标,并将其输出到外部系统。

           一般来说,您只需要去实现 LoggingMetricsConsumer,统计将指标值输出到metric.log日志文件之中。

当然,您也可以自定义一个监听的类:只需要去实现IMetricsConsumer接口就可以了。这些类可以在代码里注册(registerMetricsConsumer),也可以在 storm.yaml配置文件中注册:



© 著作权归作者所有

止静
粉丝 121
博文 134
码字总数 125762
作品 0
东城
技术主管
私信 提问
Storm【实践系列-如何写一个爬虫-】6 URLInjector

介绍: URLInjector,我封装了的一个简单的客户端,我们将要放URLs,到一个分片队列里面 【sharded queue】,只有放置到分片队列的数据才会被Storm的管线所处理。 package com.digitalpebbl...

止静
2014/08/21
110
0
Storm【实践系列-如何写一个爬虫】 - ParserBolt

阅读背景: 如果您对爬虫,或则web前端不够了解,请自行google。 代码前提:您需要参阅本ID 所写的前面两篇博文: Storm【实践系列-如何写一个爬虫】 - Fetcher 本章主题: ParserBolt 如何完...

止静
2014/08/18
1K
0
Storm入门 第三章 Storm安装部署步骤

本文以Twitter Storm官方Wiki为基础,详细描述如何快速搭建一个Storm集群,其中,项目实践中遇到的问题及经验总结,在相应章节以“注意事项”的形式给出。 3.1 Storm集群组件 Storm集群中包含...

坏坏一笑
2014/12/03
86
0
Storm【实践系列-如何写一个爬虫4】 - ParserBolt

代码精要: 1 : 首先,在新版本Storm0.92的功能中加入了Storm Metric的度量 2 :其次:在真个抽象的过程中,把具体的业务处理过程抽象为handler,其中包括了: 2.1:LinkContentHandler 2....

止静
2014/08/21
127
0
Apache Slider + Storm

Apache Slider + Storm 系统环境 安装如下组件,部署可用环境 JDK 1.7.0_79 Apache Zookeeper 3.4.* Apache Zookeeper Apache Hadoop 2.6.* Apache Hadoop Apache Storm 0.9.4 Apache Storm......

Yulong_
2016/09/21
458
0

没有更多内容

加载失败,请刷新页面

加载更多

Docker 可视化管理 portainer

官网安装指南: https://portainer.readthedocs.io/en/latest/deployment.html docker-compose.yml 位置,下载地址:https://downloads.portainer.io/docker-compose.yml...

Moks角木
43分钟前
5
0
Spring Security 实战干货:必须掌握的一些内置 Filter

1. 前言 上一文我们使用 Spring Security 实现了各种登录聚合的场面。其中我们是通过在 UsernamePasswordAuthenticationFilter 之前一个自定义的过滤器实现的。我怎么知道自定义过滤器要加在...

码农小胖哥
46分钟前
7
0
常见分布式事务解决方案

1 微服务的发展 微服务倡导将复杂的单体应用拆分为若干个功能简单、松耦合的服务,这样可以降低开发难度、增强扩展性、便于敏捷开发。当前被越来越多的开发者推崇,很多互联网行业巨头、开源...

asdf08442a
47分钟前
3
0
influxdb continuous queries(cq)从入门到放弃

从前一篇influxdb的文章prometheus基于influxdb的监控数据持久化存储方案完成之后,就一直在折腾influxdb发布测试和生产环境的问题,经过接近2个月的验证,最终发现使用influxdb自带cq的方案...

狗陈
58分钟前
7
0
7.线程通信

在现实生活中,如果一个人团队正在共同完成任务,那么他们之间应该有通信,以便正确完成任务。 同样的比喻也适用于线程。 在编程中,要减少处理器的理想时间,我们创建了多个线程,并为每个线...

Eappo_Geng
今天
6
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部