Storm的wordCounter计数器详解

原文：http://www.maoxiangyi.cn/index.php/archives/362

拓扑

点击(此处)折叠或打开

 package cn.jd.storm;

 import backtype.storm.Config;
 import backtype.storm.LocalCluster;
 import backtype.storm.topology.TopologyBuilder;
 import backtype.storm.tuple.Fields;
 /**
  * 功能说明：
  * 设计一个topology，来实现对一个句子里面的单词出现的频率进行统计。
  * 整个topology分为三个部分：
  *         WordReader:数据源，负责发送单行文本记录（句子）
  *         WordNormalizer:负责将单行文本记录（句子）切分红单词
  *         WordCounter:负责对单词的频率进行累加
  *
  * @author 毛祥溢
  * Email:frank@maoxiangyi.cn
  * 2013-8-26 下午5:59:06
  */
 public class TopologyMain {

     /**
      * @param args 文件路径
      */
     public static void main(String[] args)throws Exception {
         // Storm框架支持多语言，在JAVA环境下建立一个拓扑，须要使用TopologyBuilder进行构建
         TopologyBuilder builder = new TopologyBuilder();
         /* WordReader类，主要是将文本内容读成一行一行的模式
          * 消息源spout是Storm里面一个topology里面的消息生产者。
          * 通常来讲消息源会从一个外部源读取数据而且向topology里面发出消息：tuple。
          * Spout能够是可靠的也能够是不可靠的。
          * 若是这个tuple没有被storm成功处理，可靠的消息源spouts能够从新发射一个tuple，可是不可靠的消息源spouts一旦发出一个tuple就不能重发了。
          *
          * 消息源能够发射多条消息流stream。多条消息流能够理解为多中类型的数据。
          * 使用OutputFieldsDeclarer.declareStream来定义多个stream，而后使用SpoutOutputCollector来发射指定的stream。
          *
          * Spout类里面最重要的方法是nextTuple。要么发射一个新的tuple到topology里面或者简单的返回若是已经没有新的tuple。
          * 要注意的是nextTuple方法不能阻塞，由于storm在同一个线程上面调用全部消息源spout的方法。
          *
          * 另外两个比较重要的spout方法是ack和fail。storm在检测到一个tuple被整个topology成功处理的时候调用ack，不然调用fail。storm只对可靠的spout调用ack和fail。
          */
         builder.setSpout("word-reader",new WordReader());
         /* WordNormalizer类，主要是将一行一行的文本内容切割成单词
          *
          * 全部的消息处理逻辑被封装在bolts里面。Bolts能够作不少事情：过滤，聚合，查询数据库等等。
          * Bolts能够简单的作消息流的传递。复杂的消息流处理每每须要不少步骤，从而也就须要通过不少bolts。
          * 好比算出一堆图片里面被转发最多的图片就至少须要两步：
          *     第一步算出每一个图片的转发数量。
          *     第二步找出转发最多的前10个图片。（若是要把这个过程作得更具备扩展性那么可能须要更多的步骤）。
          *
          * Bolts能够发射多条消息流， 使用OutputFieldsDeclarer.declareStream定义stream，使用OutputCollector.emit来选择要发射的stream。
          * Bolts的主要方法是execute, 它以一个tuple做为输入，bolts使用OutputCollector来发射tuple。
          * bolts必需要为它处理的每个tuple调用OutputCollector的ack方法，以通知Storm这个tuple被处理完成了，从而通知这个tuple的发射者spouts。
          * 通常的流程是： bolts处理一个输入tuple, 发射0个或者多个tuple, 而后调用ack通知storm本身已经处理过这个tuple了。storm提供了一个IBasicBolt会自动调用ack。
          *
          *
          */
         builder.setBolt("word-normalizer", new WordNormalizer()).shuffleGrouping("word-reader");
         /*
          * 上面的代码和下面的代码中都设定了数据分配的策略stream grouping
          * 定义一个topology的其中一步是定义每一个bolt接收什么样的流做为输入。stream grouping就是用来定义一个stream应该若是分配数据给bolts上面的多个tasks。
          * Storm里面有7种类型的stream grouping
          *         Shuffle Grouping: 随机分组， 随机派发stream里面的tuple，保证每一个bolt接收到的tuple数目大体相同。
          *         Fields Grouping：按字段分组， 好比按userid来分组， 具备一样userid的tuple会被分到相同的Bolts里的一个task，
          *             而不一样的userid则会被分配到不一样的bolts里的task。
          *         All Grouping：广播发送，对于每个tuple，全部的bolts都会收到。
          *         Global Grouping：全局分组， 这个tuple被分配到storm中的一个bolt的其中一个task。再具体一点就是分配给id值最低的那个task。
          *         Non Grouping：不分组，这stream grouping个分组的意思是说stream不关心到底谁会收到它的tuple。
          *             目前这种分组和Shuffle grouping是同样的效果， 有一点不一样的是storm会把这个bolt放到这个bolt的订阅者同一个线程里面去执行。
          *         Direct Grouping： 直接分组， 这是一种比较特别的分组方法，用这种分组意味着消息的发送者指定由消息接收者的哪一个task处理这个消息。
          *             只有被声明为Direct Stream的消息流能够声明这种分组方法。并且这种消息tuple必须使用emitDirect方法来发射。
          *             消息处理者能够经过TopologyContext来获取处理它的消息的task的id （OutputCollector.emit方法也会返回task的id）。
          *         Local or shuffle grouping：若是目标bolt有一个或者多个task在同一个工做进程中，tuple将会被随机发生给这些tasks。
          *             不然，和普通的Shuffle Grouping行为一致。
          *
          */
         builder.setBolt("word-counter", new WordCounter(),1).fieldsGrouping("word-normalizer", new Fields("word"));

         /*
          * storm的运行有两种模式: 本地模式和分布式模式.
          * 1) 本地模式：
          *         storm用一个进程里面的线程来模拟全部的spout和bolt. 本地模式对开发和测试来讲比较有用。
          *         你运行storm-starter里面的topology的时候它们就是以本地模式运行的， 你能够看到topology里面的每个组件在发射什么消息。
          * 2) 分布式模式：
          *         storm由一堆机器组成。当你提交topology给master的时候， 你同时也把topology的代码提交了。
          *         master负责分发你的代码而且负责给你的topolgoy分配工做进程。若是一个工做进程挂掉了， master节点会把认为从新分配到其它节点。
          * 下面是以本地模式运行的代码:
          *
          * Conf对象能够配置不少东西， 下面两个是最多见的：
          *        TOPOLOGY_WORKERS(setNumWorkers) 定义你但愿集群分配多少个工做进程给你来执行这个topology.
          *            topology里面的每一个组件会被须要线程来执行。每一个组件到底用多少个线程是经过setBolt和setSpout来指定的。
          *            这些线程都运行在工做进程里面. 每个工做进程包含一些节点的一些工做线程。
          *            好比， 若是你指定300个线程，60个进程， 那么每一个工做进程里面要执行6个线程， 而这6个线程可能属于不一样的组件(Spout, Bolt)。
          *            你能够经过调整每一个组件的并行度以及这些线程所在的进程数量来调整topology的性能。
          *        TOPOLOGY_DEBUG(setDebug), 当它被设置成true的话， storm会记录下每一个组件所发射的每条消息。
          *            这在本地环境调试topology颇有用， 可是在线上这么作的话会影响性能的。
          */
         Config conf = new Config();
         conf.setDebug(true);
         conf.setNumWorkers(2);
         conf.put("wordsFile","/root/workspace1/com.jd.storm.demo/src/main/resources/words.txt");
         conf.setDebug(true);
         conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
         /*
          * 定义一个LocalCluster对象来定义一个进程内的集群。提交topology给这个虚拟的集群和提交topology给分布式集群是同样的。
          * 经过调用submitTopology方法来提交topology， 它接受三个参数：要运行的topology的名字，一个配置对象以及要运行的topology自己。
          * topology的名字是用来惟一区别一个topology的，这样你而后能够用这个名字来杀死这个topology的。前面已经说过了， 你必须显式的杀掉一个topology， 不然它会一直运行。
          */
         LocalCluster cluster = new LocalCluster();
         cluster.submitTopology("wordCounterTopology", conf, builder.createTopology());
         Thread.sleep(1000);
         cluster.killTopology("wordCounterTopology");
         cluster.shutdown();
     }

 }

读句子

点击(此处)折叠或打开

 package cn.jd.storm;

 import java.io.BufferedReader;
 import java.io.FileNotFoundException;
 import java.io.FileReader;
 import java.util.Map;

 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.Fields;
 import backtype.storm.tuple.Values;
 /**
  *
  * 功能说明：
  *         主要是将文件内容读出来，一行一行
  *
  *     Spout类里面最重要的方法是nextTuple。
  *     要么发射一个新的tuple到topology里面或者简单的返回若是已经没有新的tuple。
  *     要注意的是nextTuple方法不能阻塞，由于storm在同一个线程上面调用全部消息源spout的方法。
  *     另外两个比较重要的spout方法是ack和fail。
  *     storm在检测到一个tuple被整个topology成功处理的时候调用ack，不然调用fail。
  * storm只对可靠的spout调用ack和fail。
  *
  * @author 毛祥溢
  * Email:frank@maoxiangyi.cn
  * 2013-8-26 下午6:05:46
  */
 public class WordReader extends BaseRichSpout {

     private SpoutOutputCollector collector;
     private FileReader fileReader;
     private String filePath;
     private boolean completed = false;
     //storm在检测到一个tuple被整个topology成功处理的时候调用ack，不然调用fail。
     public void ack(Object msgId) {
         System.out.println("OK:"+msgId);
     }
     public void close() {}
     //storm在检测到一个tuple被整个topology成功处理的时候调用ack，不然调用fail。
     public void fail(Object msgId) {
         System.out.println("FAIL:"+msgId);
     }

   /*
    * 在SpoutTracker类中被调用，每调用一次就能够向storm集群中发射一条数据（一个tuple元组），该方法会被不停的调用
    */
     public void nextTuple() {
         if(completed){
             try {
                 Thread.sleep(1000);
             } catch (InterruptedException e) {
             }
             return;
         }
         String str;
         BufferedReader reader =new BufferedReader(fileReader);
         try{
             while((str = reader.readLine()) != null){
                 System.out.println("WordReader类 读取到一行数据："+ str);
                 this.collector.emit(new Values(str),str);
                 System.out.println("WordReader类 发射了一条数据："+ str);
             }
         }catch(Exception e){
             throw new RuntimeException("Error reading tuple",e);
         }finally{
             completed = true;
         }
     }

     public void open(Map conf, TopologyContext context,SpoutOutputCollector collector) {
         try {
             this.fileReader = new FileReader(conf.get("wordsFile").toString());
         } catch (FileNotFoundException e) {
             throw new RuntimeException("Error reading file ["+conf.get("wordFile")+"]");
         }
         this.filePath    = conf.get("wordsFile").toString();
         this.collector = collector;
     }
     /**
      * 定义字段id，该id在简单模式下没有用处，但在按照字段分组的模式下有很大的用处。
      * 该declarer变量有很大做用，咱们还能够调用declarer.declareStream();来定义stramId，该id能够用来定义更加复杂的流拓扑结构
      */
     public void declareOutputFields(OutputFieldsDeclarer declarer) {
         declarer.declare(new Fields("line"));
     }
 }

将句子切割成单词

点击(此处)折叠或打开

 package cn.jd.storm;

 import backtype.storm.topology.BasicOutputCollector;
 import backtype.storm.topology.OutputFieldsDeclarer;
 import backtype.storm.topology.base.BaseBasicBolt;
 import backtype.storm.tuple.Fields;
 import backtype.storm.tuple.Tuple;
 import backtype.storm.tuple.Values;
 /**
  *
  * 功能说明：
  *     将一行文本切割成单词，并封装collector中发射出去
  *
  * @author 毛祥溢
  * Email:frank@maoxiangyi.cn
  * 2013-8-26 下午6:05:59
  */
 public class WordNormalizer extends BaseBasicBolt {

     public void cleanup() {
         System.out.println("将一行文本切割成单词，并封装collector中发射出去 ---完毕！");
     }

     /**
      * 接受的参数是WordReader发出的句子，即input的内容是句子
      * execute方法，将句子切割造成的单词发出
      */
     public void execute(Tuple input, BasicOutputCollector collector) {
         String sentence = input.getString(0);
         String[] words = sentence.split(" ");
         System.out.println("WordNormalizer类 收到一条数据，这条数据是： "+ sentence);
         for(String word : words){
             word = word.trim();
             if(!word.isEmpty()){
                 word = word.toLowerCase();
                 System.out.println("WordNormalizer类 收到一条数据，这条数据是： "+ sentence+"数据正在被切割，切割出来的单词是 "+ word);
                 collector.emit(new Values(word));
             }
         }
     }

     /**
     * 定义字段id，该id在简单模式下没有用处，但在按照字段分组的模式下有很大的用处。
     * 该declarer变量有很大做用，咱们还能够调用declarer.declareStream();来定义stramId，该id能够用来定义更加复杂的流拓扑结构
     */
     public void declareOutputFields(OutputFieldsDeclarer declarer) {
         declarer.declare(new Fields("word"));
     }
 }

计数器WordCounter

点击(此处)折叠或打开

 package cn.jd.storm;

 import java.util.HashMap;
 import java.util.Map;

 import backtype.storm.task.TopologyContext;
 import backtype.storm.topology.BasicOutputCollector;
 import backtype.storm.topology.OutputFieldsDeclarer;
 import backtype.storm.topology.base.BaseBasicBolt;
 import backtype.storm.tuple.Tuple;
 /**
  *
  * 功能说明：
  *     实现计数器的功能，第一次将collector中的元素存放在成员变量counters（Map）中.
  *     若是counters（Map）中已经存在该元素，getValule并对Value进行累加操做。
  *
  * @author 毛祥溢
  * Email:frank@maoxiangyi.cn
  * 2013-8-26 下午6:06:07
  */
 public class WordCounter extends BaseBasicBolt {

     private static final long serialVersionUID = 5678586644899822142L;
     Integer id;
     String name;
      //定义Map封装最后的结果
     Map<String, Integer> counters;

     /**
      * 在spout结束时被调用，将最后的结果显示出来
      *
      * 結果:
      * -- Word Counter [word-counter-2] --
      * really: 1
      * but: 1
      * application: 1
      * is: 2
      * great: 2
      */
     @Override
     public void cleanup() {
         System.out.println("-- Word Counter ["+name+"-"+id+"] --");
         for(Map.Entry<String, Integer> entry : counters.entrySet()){
             System.out.println(entry.getKey()+": "+entry.getValue());
         }
         System.out.println("实现计数器的功能 --完畢！");
     }

     /**
      * 初始化操做
      */
     @Override
     public void prepare(Map stormConf, TopologyContext context) {
         this.counters = new HashMap<String, Integer>();
         this.name = context.getThisComponentId();
         this.id = context.getThisTaskId();
     }

     public void declareOutputFields(OutputFieldsDeclarer declarer) {}

     /**
      * 实现计数器的功能，第一次将collector中的元素存放在成员变量counters（Map）中.
      * 若是counters（Map）中已经存在该元素，getValule并对Value进行累加操做。
      */
     public void execute(Tuple input, BasicOutputCollector collector) {
         String str = input.getString(0);
         System.out.println("WordCounter 计数器收到单词 "+ str);
         if(!counters.containsKey(str)){
             counters.put(str, 1);
         }else{
             Integer c = counters.get(str) + 1;
             counters.put(str, c);
         }
     }
 }

数据格式

storm ni great he he xi wang test haha heihei very are mao xiang yi jd

运行日志

WordReader类 发射了一条数据：storm ni great he he xi wang WordReader类 发射了一条数据：test haha heihei very WordNormalizer类 收到一条数据，这条数据是：    storm ni great he he xi wang WordNormalizer类 收到一条数据，这条数据是：    storm ni great he he xi wang数据正在被切割，切割出来的单词是 storm WordNormalizer类 收到一条数据，这条数据是：    storm ni great he he xi wang数据正在被切割，切割出来的单词是 ni WordNormalizer类 收到一条数据，这条数据是：    storm ni great he he xi wang数据正在被切割，切割出来的单词是 great WordNormalizer类 收到一条数据，这条数据是：    storm ni great he he xi wang数据正在被切割，切割出来的单词是 he WordNormalizer类 收到一条数据，这条数据是：    storm ni great he he xi wang数据正在被切割，切割出来的单词是 he WordNormalizer类 收到一条数据，这条数据是：    storm ni great he he xi wang数据正在被切割，切割出来的单词是 xi WordNormalizer类 收到一条数据，这条数据是：    storm ni great he he xi wang数据正在被切割，切割出来的单词是 wang WordCounter 计数器收到单词 storm WordCounter 计数器收到单词 ni WordCounter 计数器收到单词 great WordCounter 计数器收到单词 he WordCounter 计数器收到单词 he WordCounter 计数器收到单词 xi WordCounter 计数器收到单词 wang WordNormalizer类 收到一条数据，这条数据是：    test haha heihei very WordNormalizer类 收到一条数据，这条数据是：    test haha heihei very 数据正在被切割，切割出来的单词是 test WordNormalizer类 收到一条数据，这条数据是：    test haha heihei very 数据正在被切割，切割出来的单词是 haha WordNormalizer类 收到一条数据，这条数据是：    test haha heihei very 数据正在被切割，切割出来的单词是 heihei WordNormalizer类 收到一条数据，这条数据是：    test haha heihei very 数据正在被切割，切割出来的单词是 very WordCounter 计数器收到单词 test WordCounter 计数器收到单词 haha WordCounter 计数器收到单词 heihei WordCounter 计数器收到单词 very WordReader类 发射了一条数据：are mao xiang yi jd WordNormalizer类 收到一条数据，这条数据是：    are mao xiang yi jd WordNormalizer类 收到一条数据，这条数据是：    are mao xiang yi jd 数据正在被切割，切割出来的单词是 are WordNormalizer类 收到一条数据，这条数据是：    are mao xiang yi jd 数据正在被切割，切割出来的单词是 mao WordNormalizer类 收到一条数据，这条数据是：    are mao xiang yi jd 数据正在被切割，切割出来的单词是 xiang WordNormalizer类 收到一条数据，这条数据是：    are mao xiang yi jd 数据正在被切割，切割出来的单词是 yi WordNormalizer类 收到一条数据，这条数据是：    are mao xiang yi jd 数据正在被切割，切割出来的单词是 jd WordCounter 计数器收到单词 are WordCounter 计数器收到单词 mao WordCounter 计数器收到单词 xiang WordCounter 计数器收到单词 yi WordCounter 计数器收到单词 jd -- Word Counter [word-counter-2] -- xi: 1 test: 1 heihei: 1 haha: 1 he: 2 storm: 1 wang: 1 jd: 1 xiang: 1 great: 1 are: 1 ni: 1 yi: 1 very: 1 mao: 1 实现计数器的功能 --完畢！ 将一行文本切割成单词，并封装collector中发射出去 ---完毕！

阅读(40) | 评论(0) | 转发(0) |

0

上一篇：开源大数据利器汇总

下一篇：Linux下的strerror是否线程安全？

相关热门文章

• Linux内存管理之mmap详解...
• 惊喜巨献HP53181A HP53181A HP...
• java 虚拟机工做原理详解...
• 转季促销会HP8904A信号发生器A...
• java并发多线程总结

• http://shopceline-2013.6te.n...
• 配置HADOOP开发环境(3)-代码开...
• Zookeeper分布式集群部署...
• hadoop学习
• Hadoop学习全程记录——在Ecli...

• 个人ChinaUnix博客被锁定了，...
• 虚拟机中ubuntu无线链接问题...
• IBM DS3400 盘阵怎么查看是单...
• 启动auditd时，报错以下，怎么...
• CGSL系统中root密码正确，但无...

给主人留下些什么吧！~~

评论热议