Chinaunix首页 | 论坛 | 博客
  • 博客访问: 3228389
  • 博文数量: 346
  • 博客积分: 10189
  • 博客等级: 上将
  • 技术积分: 3125
  • 用 户 组: 普通用户
  • 注册时间: 2008-08-05 19:46
文章分类

全部博文(346)

文章存档

2013年(35)

2011年(35)

2010年(76)

2009年(48)

2008年(152)

分类: 大数据

2013-09-13 09:14:05

转自:http://www.cnblogs.com/panfeng412/archive/2012/06/19/storm-common-patterns-of-basicbolt.html

Storm中的很多Bolt都有一个最常见的处理步骤:

  1. 读入一个tuple;
  2. 根据这个输入tuple,提取后发射0个,1个或多个tuple;
  3. 最后,通过ack操作确认这个tuple被成功处理。

按照上述处理步骤,依次处理发向这个Bolt的各个tuple元组。

这种模式可以实现像ETL这类的简单函数或过滤器功能,Storm中专门为这种模式封装了相应接口:。等类实现了这一接口。

下面是以为基础,按照上述模式实现词频统计的Bolt(代码参考链接:):

复制代码
public static class WordCount extends BaseBasicBolt {
     //记录每个单词及单词出现的次数 Map counts = new HashMap();

        @Override public void execute(Tuple tuple, BasicOutputCollector collector) {
            String word = tuple.getString(0);
            Integer count = counts.get(word); //提取单词出现次数 if(count==null)
         count = 0;
            count++;
            counts.put(word, count); //更新单词出现次数 collector.emit(new Values(word, count)); //发射统计结果  }

        @Override public void declareOutputFields(OutputFieldsDeclarer declarer) {
       declarer.declare(new Fields("word", "count"));
        }
    }
复制代码
阅读(2931) | 评论(0) | 转发(0) |
给主人留下些什么吧!~~