Storm实战之WordCount

标签： storm wordcount | 发表时间：2015-06-22 02:39 | 作者：m635674608
出处：http://www.iteye.com
 在全面介绍Storm之前，我们先通过一个简单的Demo让大家整体感受一下什么是Storm。

Storm运行模式：

  1. 本地模式(Local Mode)： 即Topology（相当于一个任务，后续会详细讲解）  运行在本地机器的单一JVM上，这个模式主要用来开发、调试。
  2. 远程模式(Remote Mode):在这个模式，我们把我们的Topology提交到集群，在这个模式中，Storm的所有组件都是线程安全的，因为它们都会运行在不同的Jvm或物理机器上，这个模式就是正式的生产模式。
写一个HelloWord Storm
     我们现在创建这么一个应用，统计文本文件中的单词个数，详细学习过Hadoop的朋友都应该写过。那么我们需要具体创建这样一个Topology，用一个spout负责读取文本文件，用第一个bolt来解析成单词，用第二个bolt来对解析出的单词计数，整体结构如图所示：
     写一个可运行的Demo很简单，我们只需要三步：
  1. 创建一个Spout读取数据
  2. 创建bolt处理数据
  3. 创建一个Topology提交到集群
下面我们就写一下，以下代码拷贝到eclipse（依赖的jar包到官网下载即可）即可运行。
1.创建一个Spout作为数据源
     Spout作为数据源，它实现了IRichSpout接口，功能是读取一个文本文件并把它的每一行内容发送给bolt。
[java]  view plain copy
 
  1. package storm.demo.spout;  
  2.   
  3. import java.io.BufferedReader;  
  4. import java.io.FileNotFoundException;  
  5. import java.io.FileReader;  
  6. import java.util.Map;  
  7. import backtype.storm.spout.SpoutOutputCollector;  
  8. import backtype.storm.task.TopologyContext;  
  9. import backtype.storm.topology.IRichSpout;  
  10. import backtype.storm.topology.OutputFieldsDeclarer;  
  11. import backtype.storm.tuple.Fields;  
  12. import backtype.storm.tuple.Values;  
  13. public class WordReader implements IRichSpout {  
  14.     private static final long serialVersionUID = 1L;  
  15.     private SpoutOutputCollector collector;  
  16.     private FileReader fileReader;  
  17.     private boolean completed = false;  
  18.   
  19.     public boolean isDistributed() {  
  20.         return false;  
  21.     }  
  22.     /** 
  23.      * 这是第一个方法，里面接收了三个参数，第一个是创建Topology时的配置， 
  24.      * 第二个是所有的Topology数据，第三个是用来把Spout的数据发射给bolt 
  25.      * **/  
  26.     @Override  
  27.     public void open(Map conf, TopologyContext context,  
  28.             SpoutOutputCollector collector) {  
  29.         try {  
  30.             //获取创建Topology时指定的要读取的文件路径  
  31.             this.fileReader = new FileReader(conf.get("wordsFile").toString());  
  32.         } catch (FileNotFoundException e) {  
  33.             throw new RuntimeException("Error reading file ["  
  34.                     + conf.get("wordFile") + "]");  
  35.         }  
  36.         //初始化发射器  
  37.         this.collector = collector;  
  38.   
  39.     }  
  40.     /** 
  41.      * 这是Spout最主要的方法，在这里我们读取文本文件，并把它的每一行发射出去（给bolt） 
  42.      * 这个方法会不断被调用，为了降低它对CPU的消耗，当任务完成时让它sleep一下 
  43.      * **/  
  44.     @Override  
  45.     public void nextTuple() {  
  46.         if (completed) {  
  47.             try {  
  48.                 Thread.sleep(1000);  
  49.             } catch (InterruptedException e) {  
  50.                 // Do nothing  
  51.             }  
  52.             return;  
  53.         }  
  54.         String str;  
  55.         // Open the reader  
  56.         BufferedReader reader = new BufferedReader(fileReader);  
  57.         try {  
  58.             // Read all lines  
  59.             while ((str = reader.readLine()) != null) {  
  60.                 /** 
  61.                  * 发射每一行，Values是一个ArrayList的实现 
  62.                  */  
  63.                 this.collector.emit(new Values(str), str);  
  64.             }  
  65.         } catch (Exception e) {  
  66.             throw new RuntimeException("Error reading tuple", e);  
  67.         } finally {  
  68.             completed = true;  
  69.         }  
  70.   
  71.     }  
  72.     @Override  
  73.     public void declareOutputFields(OutputFieldsDeclarer declarer) {  
  74.         declarer.declare(new Fields("line"));  
  75.   
  76.     }  
  77.     @Override  
  78.     public void close() {  
  79.         // TODO Auto-generated method stub  
  80.     }  
  81.       
  82.     @Override  
  83.     public void activate() {  
  84.         // TODO Auto-generated method stub  
  85.   
  86.     }  
  87.     @Override  
  88.     public void deactivate() {  
  89.         // TODO Auto-generated method stub  
  90.   
  91.     }  
  92.     @Override  
  93.     public void ack(Object msgId) {  
  94.         System.out.println("OK:" + msgId);  
  95.     }  
  96.     @Override  
  97.     public void fail(Object msgId) {  
  98.         System.out.println("FAIL:" + msgId);  
  99.   
  100.     }  
  101.     @Override  
  102.     public Map<String, Object> getComponentConfiguration() {  
  103.         // TODO Auto-generated method stub  
  104.         return null;  
  105.     }  
  106. }  
2.创建两个bolt来处理Spout发射出的数据
     Spout已经成功读取文件并把每一行作为一个tuple(在Storm数据以tuple的形式传递)发射过来，我们这里需要创建两个bolt分别来负责解析每一行和对单词计数。
     Bolt中最重要的是execute方法，每当一个tuple传过来时它便会被调用。
     第一个bolt：WordNormalizer
[java]  view plain copy
 
  1. package storm.demo.bolt;  
  2. import java.util.ArrayList;  
  3. import java.util.List;  
  4. import java.util.Map;  
  5. import backtype.storm.task.OutputCollector;  
  6. import backtype.storm.task.TopologyContext;  
  7. import backtype.storm.topology.IRichBolt;  
  8. import backtype.storm.topology.OutputFieldsDeclarer;  
  9. import backtype.storm.tuple.Fields;  
  10. import backtype.storm.tuple.Tuple;  
  11. import backtype.storm.tuple.Values;  
  12. public class WordNormalizer implements IRichBolt {  
  13.     private OutputCollector collector;  
  14.     @Override  
  15.     public void prepare(Map stormConf, TopologyContext context,  
  16.             OutputCollector collector) {  
  17.         this.collector = collector;  
  18.     }  
  19.     /**这是bolt中最重要的方法，每当接收到一个tuple时，此方法便被调用 
  20.      * 这个方法的作用就是把文本文件中的每一行切分成一个个单词，并把这些单词发射出去（给下一个bolt处理） 
  21.      * **/  
  22.     @Override  
  23.     public void execute(Tuple input) {  
  24.         String sentence = input.getString(0);  
  25.         String[] words = sentence.split(" ");  
  26.         for (String word : words) {  
  27.             word = word.trim();  
  28.             if (!word.isEmpty()) {  
  29.                 word = word.toLowerCase();  
  30.                 // Emit the word  
  31.                 List a = new ArrayList();  
  32.                 a.add(input);  
  33.                 collector.emit(a, new Values(word));  
  34.             }  
  35.         }  
  36.         //确认成功处理一个tuple  
  37.         collector.ack(input);  
  38.     }  
  39.     @Override  
  40.     public void declareOutputFields(OutputFieldsDeclarer declarer) {  
  41.         declarer.declare(new Fields("word"));  
  42.   
  43.     }  
  44.     @Override  
  45.     public void cleanup() {  
  46.         // TODO Auto-generated method stub  
  47.   
  48.     }  
  49.     @Override  
  50.     public Map<String, Object> getComponentConfiguration() {  
  51.         // TODO Auto-generated method stub  
  52.         return null;  
  53.     }  
  54. }  
     第二个bolt:WordCounter
[java]  view plain copy
 
  1. package storm.demo.bolt;  
  2. import java.util.HashMap;  
  3. import java.util.Map;  
  4. import backtype.storm.task.OutputCollector;  
  5. import backtype.storm.task.TopologyContext;  
  6. import backtype.storm.topology.IRichBolt;  
  7. import backtype.storm.topology.OutputFieldsDeclarer;  
  8. import backtype.storm.tuple.Tuple;  
  9.   
  10. public class WordCounter implements IRichBolt {  
  11.     Integer id;  
  12.     String name;  
  13.     Map<String, Integer> counters;  
  14.     private OutputCollector collector;  
  15.   
  16.     @Override  
  17.     public void prepare(Map stormConf, TopologyContext context,  
  18.             OutputCollector collector) {  
  19.         this.counters = new HashMap<String, Integer>();  
  20.         this.collector = collector;  
  21.         this.name = context.getThisComponentId();  
  22.         this.id = context.getThisTaskId();  
  23.   
  24.     }  
  25.     @Override  
  26.     public void execute(Tuple input) {  
  27.         String str = input.getString(0);  
  28.         if (!counters.containsKey(str)) {  
  29.             counters.put(str, 1);  
  30.         } else {  
  31.             Integer c = counters.get(str) + 1;  
  32.             counters.put(str, c);  
  33.         }  
  34.         // 确认成功处理一个tuple  
  35.         collector.ack(input);  
  36.     }  
  37.     /** 
  38.      * Topology执行完毕的清理工作，比如关闭连接、释放资源等操作都会写在这里 
  39.      * 因为这只是个Demo，我们用它来打印我们的计数器 
  40.      * */  
  41.     @Override  
  42.     public void cleanup() {  
  43.         System.out.println("-- Word Counter [" + name + "-" + id + "] --");  
  44.         for (Map.Entry<String, Integer> entry : counters.entrySet()) {  
  45.             System.out.println(entry.getKey() + ": " + entry.getValue());  
  46.         }  
  47.         counters.clear();  
  48.     }  
  49.     @Override  
  50.     public void declareOutputFields(OutputFieldsDeclarer declarer) {  
  51.         // TODO Auto-generated method stub  
  52.   
  53.     }  
  54.     @Override  
  55.     public Map<String, Object> getComponentConfiguration() {  
  56.         // TODO Auto-generated method stub  
  57.         return null;  
  58.     }  
  59. }  
3.在main函数中创建一个Topology
     在这里我们要创建一个Topology和一个LocalCluster对象，还有一个Config对象做一些配置。   
[java]  view plain copy
 
  1. package storm.demo;  
  2.   
  3. import storm.demo.bolt.WordCounter;  
  4. import storm.demo.bolt.WordNormalizer;  
  5. import storm.demo.spout.WordReader;  
  6. import backtype.storm.Config;  
  7. import backtype.storm.LocalCluster;  
  8. import backtype.storm.topology.TopologyBuilder;  
  9. import backtype.storm.tuple.Fields;  
  10. public class WordCountTopologyMain {  
  11.     public static void main(String[] args) throws InterruptedException {  
  12.         //定义一个Topology  
  13.         TopologyBuilder builder = new TopologyBuilder();  
  14.         builder.setSpout("word-reader",new WordReader());  
  15.         builder.setBolt("word-normalizer", new WordNormalizer())  
  16.         .shuffleGrouping("word-reader");  
  17.         builder.setBolt("word-counter", new WordCounter(),2)  
  18.         .fieldsGrouping("word-normalizer", new Fields("word"));  
  19.         //配置  
  20.         Config conf = new Config();  
  21.         conf.put("wordsFile", "d:/text.txt");  
  22.         conf.setDebug(false);  
  23.         //提交Topology  
  24.         conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);  
  25.         //创建一个本地模式cluster  
  26.         LocalCluster cluster = new LocalCluster();  
  27.         cluster.submitTopology("Getting-Started-Toplogie", conf,  
  28.         builder.createTopology());  
  29.         Thread.sleep(1000);  
  30.         cluster.shutdown();  
  31.     }  

 

 

 

 

http://fireinwind.iteye.com/blog/2153699

  1. }  
     运行这个函数我们即可看到后台打印出来的单词个数。
    （ps:因为是Local模式，运行开始可能会打印很多错误log，这个先不用管）
 


已有 0 人发表留言，猛击->> 这里<<-参与讨论


ITeye推荐



相关 [storm wordcount] 推荐：

Storm实战之WordCount

- - 编程语言 - ITeye博客
 在全面介绍Storm之前，我们先通过一个简单的Demo让大家整体感受一下什么是Storm. 本地模式(Local Mode)： 即Topology（相当于一个任务，后续会详细讲解）  运行在本地机器的单一JVM上，这个模式主要用来开发、调试. 远程模式(Remote Mode):在这个模式，我们把我们的Topology提交到集群，在这个模式中，Storm的所有组件都是线程安全的，因为它们都会运行在不同的Jvm或物理机器上，这个模式就是正式的生产模式.

wordcount详解（借鉴）

- - CSDN博客云计算推荐文章
1、MapReduce理论简介. 1.1 MapReduce编程模型. 　　MapReduce采用"分而治之"的思想，把对大规模数据集的操作，分发给一个主节点管理下的各个分节点共同完成，然后通过整合各个节点的中间结果，得到最终结果. 简单地说，MapReduce就是"任务的分解与结果的汇总". 　　在Hadoop中，用于执行MapReduce任务的机器角色有两个：一个是JobTracker；另一个是TaskTracker，JobTracker是用于调度工作的，TaskTracker是用于执行工作的.

Eclipse执行Hadoop WordCount

- - CSDN博客云计算推荐文章
Eclipse执行Hadoop WordCount.   我的Eclipse是安装在windows下的，通过Eclipse执行程序连接Hadoop，需要让虚拟机的访问地址和本机的访问地址保持在同一域内，虚拟机的地址更改前面的文章介绍过了，如果想改windows本机ip地址，打开“网络和共享中心“，点击左侧菜单”更改适配器设置“，选择相应连接网络进行IpV4属性地址修改即可.

storm简介

- - 搜索技术博客－淘宝
伴随着信息科技日新月异的发展，信息呈现出爆发式的膨胀，人们获取信息的途径也更加多样、更加便捷，同时对于信息的时效性要求也越来越高. 举个搜索场景中的例子，当一个卖家发布了一条宝贝信息时，他希望的当然是这个宝贝马上就可以被卖家搜索出来、点击、购买啦，相反，如果这个宝贝要等到第二天或者更久才可以被搜出来，估计这个大哥就要骂娘了.

Storm Trident 学习

- - 小火箭
Storm支持的三种语义：. 至少一次语义的Topology写法. 参考资料： Storm消息的可靠性保障 Storm提供了Acker的机制来保证数据至少被处理一次，是由编程人员决定是否使用这一特性，要使用这一特性需要：. 在Spout emit时添加一个MsgID，那么ack和fail方法将会被调用当Tuple被正确地处理了或发生了错误.

hadoop学习（七）WordCount+Block+Split+Shuffle+Map+Reduce技术详解

- - CSDN博客数据库推荐文章
纯干活：通过WourdCount程序示例：详细讲解MapReduce之Block+Split+Shuffle+Map+Reduce的区别及数据处理流程.        Shuffle过程是MapReduce的核心，集中了MR过程最关键的部分. 要想了解MR，Shuffle是必须要理解的. 了解Shuffle的过程，更有利于我们在对MapReduce job性能调优的工作，以及对MR内部机理有更深一步的了解.

storm常见问题解答

- - BlogJava-庄周梦蝶
    最近有朋友给我邮件问一些storm的问题，集中解答在这里. 一、我有一个数据文件，或者我有一个系统里面有数据，怎么导入storm做计算. 你需要实现一个Spout，Spout负责将数据emit到storm系统里，交给bolts计算. 怎么实现spout可以参考官方的kestrel spout实现：.

Storm 实时性分析

- - CSDN博客架构设计推荐文章
都说Storm是一个实时流处理系统，但Storm的实时性体现在什么方面呢. 首先有一个前提：这里的实时性和我们通常所说的实时系统（芯片+汇编或C编写的实时处理软件）的实时性肯定是没法比的，也不是同一个概念. 这里的实时性应该是一个相对的实时性（相对于Hadoop之类 ）. 总结一下，Storm的实时性可能主要体现在：.

那些storm的坑坑

- - 开源软件 - ITeye博客
转载请声明出处：http://blackwing.iteye.com/blog/2147633. 在使用storm的过程中，感觉它还是不如hadoop那么成熟. 当然，它的流式处理能力挺让人眼前一亮，以前做的个性化推荐都是离线计算，现在总算把实时部分也加上了. 总结一下storm使用的些心得：. 1.尽量把大量数据处理行为分拆成多个处理component.

storm准实时应用

- - CSDN博客推荐文章
1 应用背景： 需要实时统计用户的登陆数，在线人数，活跃时间，下载等指标的数据，或者清洗后移到hdfs上.         1） 客户端产生数据---.         2） kafka-生产者实时采集数据（保留7天）-----.         3） storm实时消费数据，处理数据.         4）把实时数据统计结果缓存到memcached 中.