分享

Storm常见模式1——BasicBolt

hyj 发表于 2014-4-14 15:50:26 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 1 9761
本帖最后由 pig2 于 2014-4-14 16:13 编辑
1.Storm中与各个进程与hadoop进程对应关系是怎么样的?
2.Storm中与hadoop对应的mapreduce是什么?



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

读入一个tuple;
根据这个输入tuple,提取后发射0个,1个或多个tuple;
最后,通过ack操作确认这个tuple被成功处理。
按照上述处理步骤,依次处理发向这个Bolt的各个tuple元组。

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

为了能更好的理解Storm,及出现的术语,这里提供一张图:
20130728110153_883.jpg


下面是以BaseBasicBolt为基础,按照上述模式实现词频统计的Bolt(代码参考链接:storm-starter):
  1. public static class WordCount extends BaseBasicBolt {
  2.      //记录每个单词及单词出现的次数
  3.         Map<String, Integer> counts = new HashMap<String, Integer>();
  4.         @Override
  5.         public void execute(Tuple tuple, BasicOutputCollector collector) {
  6.             String word = tuple.getString(0);
  7.             Integer count = counts.get(word); //提取单词出现次数
  8.             if(count==null)
  9.          count = 0;
  10.             count++;
  11.             counts.put(word, count); //更新单词出现次数
  12.             collector.emit(new Values(word, count)); //发射统计结果
  13.         }
  14.         @Override
  15.         public void declareOutputFields(OutputFieldsDeclarer declarer) {
  16.        declarer.declare(new Fields("word", "count"));
  17.         }
  18.     }
复制代码







已有(1)人评论

跳转到指定楼层
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条