ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > 数据分析 > Storm实战之TopN

Storm实战之TopN

原创 数据分析 作者：541732025 时间：2015-04-16 14:03:15 0 删除 编辑
TopN这种统计场景很常见，例如，统计出搜索热度最高的词，点击率最高的广告等，现在有了Hadoop、Storm这些工具之后，很方便地就能得到结果。
这里以Storm为例，简易地实现了TopN单词的统计，由于刚刚入门，代码写得比较简单。
首先，在多台机器上运行多个bolt，每个bolt负责计算一部分word的TopN，最后有一个全局的bolt，合并上一步的结果，最后得出全局的TopN。

点击(此处)折叠或打开

  1. package test.storm.topology;

  3. import test.storm.bolt.WordCounter;
  4. import test.storm.bolt.WordWriter;
  5. import test.storm.spout.WordReader;
  6. import backtype.storm.Config;
  7. import backtype.storm.StormSubmitter;
  8. import backtype.storm.generated.AlreadyAliveException;
  9. import backtype.storm.generated.InvalidTopologyException;
  10. import backtype.storm.topology.TopologyBuilder;
  11. import backtype.storm.tuple.Fields;

  13. public class WordTopN {
  14.     public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
  15.         if (args == null || args.length < 1) {
  16.             System.err.println("Usage: N");
  17.             System.err.println("such as : 10");
  18.             System.exit(-1);
  19.         }

  21.         TopologyBuilder builder = new TopologyBuilder();
  22.         builder.setSpout("wordreader", new WordReader(), 2);
  23.         builder.setBolt("wordcounter", new WordCounter(), 2).fieldsGrouping("wordreader", new Fields("word"));
  24.         builder.setBolt("wordwriter", new WordWriter()).globalGrouping("wordcounter");

  26.         Config conf = new Config();
  27.         conf.put("N", args[0]);

  29.         conf.setDebug(false);
  30.         StormSubmitter.submitTopology("topN", conf, builder.createTopology());

  32.     }
  33. }
这里需要注意的几点是，第一个bolt的分组策略是fieldsGrouping，按照字段分组，这一点很重要，它能保证相同的word被分发到同一个bolt上，
像做wordcount、TopN之类的应用就要使用这种分组策略。
最后一个bolt的分组策略是globalGrouping，全局分组，tuple会被分配到一个bolt用来汇总。
为了提高并行度，spout和第一个bolt均设置并行度为2（我这里测试机器性能不是很高）。

点击(此处)折叠或打开

  1. package test.storm.spout;

  3. import java.util.Map;
  4. import java.util.Random;
  5. import java.util.concurrent.atomic.AtomicInteger;

  7. import backtype.storm.spout.SpoutOutputCollector;
  8. import backtype.storm.task.TopologyContext;
  9. import backtype.storm.topology.OutputFieldsDeclarer;
  10. import backtype.storm.topology.base.BaseRichSpout;
  11. import backtype.storm.tuple.Fields;
  12. import backtype.storm.tuple.Values;

  14. public class WordReader extends BaseRichSpout {
  15.     private static final long serialVersionUID = 2197521792014017918L;
  16.     private SpoutOutputCollector collector;
  17.     private static AtomicInteger i = new AtomicInteger();
  18.     private static String[] words = new String[] { \"a\", \"b\", \"c\", \"d\", \"e\", \"f\", \"g\", \"h\", \"i\", \"j\", \"k\", \"l\", \"m\",
  19.             \"n\", \"o\", \"p\", \"q\", \"r\", \"s\", \"t\", \"u\", \"v\", \"w\", \"x\", \"y\", \"z\" };

  21.     @Override
  22.     public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
  23.         this.collector = collector;
  24.     }

  26.     @Override
  27.     public void nextTuple() {
  28.         if (i.intValue() < 100) {
  29.             Random rand = new Random();
  30.             String word = words[rand.nextInt(words.length)];
  31.             collector.emit(new Values(word));
  32.             i.incrementAndGet();
  33.         }
  34.     }

  36.     @Override
  37.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  38.         declarer.declare(new Fields("word"));
  39.     }
  40. }
spout的作用是随机发送word，发送100次，由于并行度是2，将产生2个spout实例，所以这里的计数器使用了static的AtomicInteger来保证线程安全。


点击(此处)折叠或打开

  1. package test.storm.bolt;

  3. import java.util.ArrayList;
  4. import java.util.Collections;
  5. import java.util.Comparator;
  6. import java.util.HashMap;
  7. import java.util.List;
  8. import java.util.Map;
  9. import java.util.Map.Entry;
  10. import java.util.concurrent.ConcurrentHashMap;

  12. import backtype.storm.task.OutputCollector;
  13. import backtype.storm.task.TopologyContext;
  14. import backtype.storm.topology.IRichBolt;
  15. import backtype.storm.topology.OutputFieldsDeclarer;
  16. import backtype.storm.tuple.Fields;
  17. import backtype.storm.tuple.Tuple;
  18. import backtype.storm.tuple.Values;

  20. public class WordCounter implements IRichBolt {
  21.     private static final long serialVersionUID = 5683648523524179434L;
  22.     private static Map<String, Integer> counters = new ConcurrentHashMap<String, Integer>();
  23.     private volatile boolean edit = true;

  25.     @Override
  26.     public void prepare(final Map stormConf, TopologyContext context, final OutputCollector collector) {
  27.         new Thread(new Runnable() {
  28.             @Override
  29.             public void run() {
  30.                 while (true) {
  31.                     //5秒后counter不再变化，可以认为spout已经发送完毕
  32.                     if (!edit) {
  33.                         if (counters.size() > 0) {
  34.                             List<Map.Entry<String, Integer>> list = new ArrayList<Map.Entry<String, Integer>>();
  35.                             list.addAll(counters.entrySet());
  36.                             Collections.sort(list, new ValueComparator());

  38.                             //向下一个bolt发送前N个word
  39.                             for (int i = 0; i < list.size(); i++) {
  40.                                 if (i < Integer.parseInt(stormConf.get("N").toString())) {
  41.                                     collector.emit(new Values(list.get(i).getKey() + ":" + list.get(i).getValue()));
  42.                                 }
  43.                             }
  44.                         }

  46.                         //发送之后，清空counters，以防spout再次发送word过来
  47.                         counters.clear();
  48.                     }

  50.                     edit = false;
  51.                     try {
  52.                         Thread.sleep(5000);
  53.                     } catch (InterruptedException e) {
  54.                         e.printStackTrace();
  55.                     }
  56.                 }
  57.             }
  58.         }).start();
  59.     }

  61.     @Override
  62.     public void execute(Tuple tuple) {
  63.         String str = tuple.getString(0);
  64.         if (counters.containsKey(str)) {
  65.             Integer c = counters.get(str) + 1;
  66.             counters.put(str, c);
  67.         } else {
  68.             counters.put(str, 1);
  69.         }

  71.         edit = true;
  72.     }

  74.     private static class ValueComparator implements Comparator<Map.Entry<String, Integer>> {
  75.         @Override
  76.         public int compare(Entry<String, Integer> entry1, Entry<String, Integer> entry2) {
  77.             return entry2.getValue() - entry1.getValue();
  78.         }
  79.     }

  81.     @Override
  82.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  83.         declarer.declare(new Fields("word_count"));
  84.     }

  86.     @Override
  87.     public void cleanup() {
  88.     }

  90.     @Override
  91.     public Map<String, Object> getComponentConfiguration() {
  92.         return null;
  93.     }
  94. }
在WordCounter里面有个线程安全的容器ConcurrentHashMap，来存储word以及对应的次数。在prepare方法里启动一个线程，长期监听edit的状态，监听间隔是5秒，
当edit为false，即execute方法不再执行、容器不再变化，可以认为spout已经发送完毕了，可以开始排序取TopN了。这里使用了一个volatile edit（回忆一下volatile的使用场景：
对变量的修改不依赖变量当前的值，这里设置true or false，显然不相互依赖）。


点击(此处)折叠或打开

  1. package test.storm.bolt;

  3. import java.io.FileWriter;
  4. import java.io.IOException;
  5. import java.util.Map;

  7. import backtype.storm.task.TopologyContext;
  8. import backtype.storm.topology.BasicOutputCollector;
  9. import backtype.storm.topology.OutputFieldsDeclarer;
  10. import backtype.storm.topology.base.BaseBasicBolt;
  11. import backtype.storm.tuple.Tuple;

  13. public class WordWriter extends BaseBasicBolt {
  14.     private static final long serialVersionUID = -6586283337287975719L;
  15.     private FileWriter writer = null;

  17.     public WordWriter() {
  18.     }

  20.     @Override
  21.     public void prepare(Map stormConf, TopologyContext context) {
  22.         try {
  23.             writer = new FileWriter("/data/tianzhen/output/" + this);
  24.         } catch (IOException e) {
  25.             e.printStackTrace();
  26.         }
  27.     }

  29.     @Override
  30.     public void execute(Tuple input, BasicOutputCollector collector) {
  31.         String s = input.getString(0);
  32.         try {
  33.             writer.write(s);
  34.             writer.write("\n");
  35.             writer.flush();
  36.         } catch (IOException e) {
  37.             e.printStackTrace();
  38.         } finally {
  39.             //writer不能close，因为execute需要一直运行
  40.         }
  41.     }

  43.     @Override
  44.     public void declareOutputFields(OutputFieldsDeclarer declarer) {

  46.     }
  47. }
最后一个bolt做全局的汇总，这里我偷了懒，直接将结果写到文件了，省略截取TopN的过程，因为我这里就一个supervisor节点，所以结果是正确的。

来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/28912557/viewspace-1579860/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： Map根据Value排序
下一篇： MySQL之timestamp
请登录后发表评论 登录
全部评论
541732025

注册时间：2013-05-23

  • 博文量
    127
  • 访问量
    516214

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us 编辑准则

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593