Storm流方式的统计系统怎么实现

本篇内容主要讲解"Storm流方式的统计系统怎么实现"，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习"Storm流方式的统计系统怎么实现"吧!

1： 初期硬件准备：

1 如果条件具备：请保证您安装好了 redis集群

2 配置好您的Storm开发环境

3 保证好您的开发环境的畅通： 主机与主机之间，Storm与redis之间

2：业务背景的介绍：

1 在这里我们将模拟一个 流方式的数据处理过程

2 数据的源头保存在我们的redis 集群之中

3 发射的数据格式为： ip，url，client_key

数据发射器

package storm.spout;import backtype.storm.spout.SpoutOutputCollector;import backtype.storm.task.TopologyContext;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.topology.base.BaseRichSpout;import backtype.storm.tuple.Values;import backtype.storm.tuple.Fields;import org.json.simple.JSONObject;import org.json.simple.JSONValue;import redis.clients.jedis.Jedis;import storm.utils.Conf;import java.util.Map;import org.apache.log4j.Logger;/** * click Spout 从redis中间读取所需要的数据 */public class ClickSpout extends BaseRichSpout {        private static final long serialVersionUID = -6200450568987812474L;        public static Logger LOG = Logger.getLogger(ClickSpout.class);        // 对于redis，我们使用的是jedis客户端        private Jedis jedis;        // 主机        private String host;        // 端口        private int port;        // Spout 收集器        private SpoutOutputCollector collector;        @Override        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {                        // 这里，我们发射的格式为                // IP，URL，CLIENT_KEY                outputFieldsDeclarer.declare(new Fields(storm.cookbook.Fields.IP,                                storm.cookbook.Fields.URL, storm.cookbook.Fields.CLIENT_KEY));        }        @Override        public void open(Map conf, TopologyContext topologyContext,                        SpoutOutputCollector spoutOutputCollector) {                host = conf.get(Conf.REDIS_HOST_KEY).toString();                port = Integer.valueOf(conf.get(Conf.REDIS_PORT_KEY).toString());                this.collector = spoutOutputCollector;                connectToRedis();        }        private void connectToRedis() {                jedis = new Jedis(host, port);        }        @Override        public void nextTuple() {                String content = jedis.rpop("count");                if (content == null || "nil".equals(content)) {                        try {                                Thread.sleep(300);                        } catch (InterruptedException e) {                        }                } else {                        // 将jedis对象 rpop出来的字符串解析为 json对象                        JSONObject obj = (JSONObject) JSONValue.parse(content);                        String ip = obj.get(storm.cookbook.Fields.IP).toString();                        String url = obj.get(storm.cookbook.Fields.URL).toString();                        String clientKey = obj.get(storm.cookbook.Fields.CLIENT_KEY)                                        .toString();                        System.out.println("this is a clientKey");                        // List