文档章节

Storm运行代码 案列

bigdataboy
 bigdataboy
发布于 2016/06/14 14:12
字数 1398
阅读 11
收藏 0

写一个HelloWord Storm

     我们现在创建这么一个应用,统计文本文件中的单词个数,详细学习过Hadoop的朋友都应该写过。那么我们需要具体创建这样一个Topology,用一个spout负责读取文本文件,用第一个bolt来解析成单词,用第二个bolt来对解析出的单词计数,整体结构如图所示:

可以从这里下载源码:https://github.com/storm-book/examples-ch02-getting_started/zipball/master

     写一个可运行的Demo很简单,我们只需要三步:

  1. 创建一个Spout读取数据
  2. 创建bolt处理数据
  3. 创建一个Topology提交到集群

下面我们就写一下,以下代码拷贝到eclipse(依赖的jar包到官网下载即可)即可运行。

1.创建一个Spout作为数据源

     Spout作为数据源,它实现了IRichSpout接口,功能是读取一个文本文件并把它的每一行内容发送给bolt。

[java] view plaincopy

  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 plaincopy

  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 plaincopy

  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 plaincopy

  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.     }  
  32. }  

     运行这个函数我们即可看到后台打印出来的单词个数。

    (ps:因为是Local模式,运行开始可能会打印很多错误log,这个先不用管)

© 著作权归作者所有

bigdataboy
粉丝 0
博文 8
码字总数 7704
作品 0
朝阳
程序员
私信 提问
storm client command

最近在研究实时日志分析,storm确实不错,以下是命令参数: storm help Syntax: storm jar topology-jar-path class 运行jar包中类的主函数和指定的参数 Commands: activate storm activate ...

China_OS
2014/02/22
0
0
Storm概念讲解和工作原理介绍

Strom的结构 Storm与传统关系型数据库 传统关系型数据库是先存后计算,而storm则是先算后存,甚至不存 传统关系型数据库很难部署实时计算,只能部署定时任务统计分析窗口数据 关系型数据库重...

张超
2015/04/26
0
0
Apache Storm 的历史及经验教训

Apache Storm 最近成为了ASF的顶级项目,这对于该项目和我个人而言是一个重大的里程碑。很难想像4年前Storm只是我脑海中的一个想法,但现在却成为了一个有着大社区支持并被无数企业使用的繁荣...

run_mei
2014/10/14
11.5K
10
windows 安装 storm 及 eclipse 调试 TopN 实例

一:安装JDK 下载地址:地址一 地址二 配置Java环境变量 JAVAHOME、Path、CLASSPATH三个值分别为(按照自己安装状况设置,此处供参考): D:javajdk1.8 %JAVAHOME%/bin;%JAVAHOME%/jre/bin ....

大数据之路
2012/06/08
0
1
【Storm】Storm简介及Storm集群的安装部署

版权声明:本文为博主原创文章,转载请注明出处。 https://blog.csdn.net/gongxifacai_believe/article/details/86557602 1、Storm概述 (1)Storm简介 Storm最早是由BackType公司开发的实时...

魏晓蕾
01/20
0
0

没有更多内容

加载失败,请刷新页面

加载更多

日志相关---日志配置和过滤器

一、log4j日志简介 1.1、 Loggers 级别和介绍 Loggers组件在此系统中被分为八个级别:ALL、TRANCE、DEBUG、INFO、WARN、ERROR和FATAL、OFF。这八个级别是有顺序的, ##off表示关闭ALL < T...

spinachgit
3分钟前
1
0
六个面试题层层剖析——LongAddr原子类

并发编程面试题 (1)LongAddr的结构是怎样的? (2)当前线程应该访问Cell数组里面的哪一个Cell元素? (3)如何初始化Cell数组? (4)Cell数组如何扩容? (5)线程访问分配的Cell元素有冲...

须臾之余
5分钟前
1
0
MySQL-入门(二)

本部分主要是MySQL的常用函数和高级用法。 一、MySQL排序 排序关键字:order by 排序字段。后面写上要排序字段,排序字段可以有多个,多个采用逗号间隔,order by默认采用升序(asc)排序,可...

潜行-L
15分钟前
0
0
BAM转VCF的方法对比

1 使用GATK HaplotypeCaller #java -jar gatk.jar HaplotypeCaller --native-pair-hmm-threads 4 -R xx.fa -I xx.bam -O xx.vcf --native-pair-hmm-threads用来设置多线程,默认为4线程 2 sa......

悲催的古灵武士
24分钟前
0
0
软件架构设计原则之“KISS”的总结使用

今天聊一聊软件架构设计中的 KISS 原则。 对! 就是亲嘴的那个 “KISS”! 一定要多练习。 ... ... ... ... 作为一个程序员我是推荐理解为“亲嘴”的,可以很好的解决单身问题,但作为一个架...

Owen_Jia
26分钟前
0
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部