内容简介:Storm详解二、写第一个Storm应用
在全面介绍Storm之前,我们先通过一个简单的Demo让大家整体感受一下什么是Storm。
Storm运行模式:
- 本地模式(Local Mode): 即Topology(相当于一个任务,后续会详细讲解) 运行在本地机器的单一JVM上,这个模式主要用来开发、调试。
- 远程模式(Remote Mode):在这个模式,我们把我们的Topology提交到集群,在这个模式中,Storm的所有组件都是线程安全的,因为它们都会运行在不同的Jvm或物理机器上,这个模式就是正式的生产模式。
写一个HelloWord Storm
我们现在创建这么一个应用,统计文本文件中的单词个数,详细学习过Hadoop的朋友都应该写过。那么我们需要具体创建这样一个Topology,用一个spout负责读取文本文件,用第一个bolt来解析成单词,用第二个bolt来对解析出的单词计数,整体结构如图所示:
可以从这里下载源码: https://github.com/storm-book/examples-ch02-getting_started/zipball/master
写一个可运行的Demo很简单,我们只需要三步:
- 创建一个Spout读取数据
- 创建bolt处理数据
- 创建一个Topology提交到集群
下面我们就写一下,以下代码拷贝到eclipse(依赖的jar包到官网下载即可)即可运行。
1.创建一个Spout作为数据源
Spout作为数据源,它实现了IRichSpout接口,功能是读取一个文本文件并把它的每一行内容发送给bolt。
package storm.demo.spout; 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.IRichSpout; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Values; public class WordReader implements IRichSpout { private static final long serialVersionUID = 1L; private SpoutOutputCollector collector; private FileReader fileReader; private boolean completed = false; public boolean isDistributed() { return false; } /** * 这是第一个方法,里面接收了三个参数,第一个是创建Topology时的配置, * 第二个是所有的Topology数据,第三个是用来把Spout的数据发射给bolt * **/ @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { try { //获取创建Topology时指定的要读取的文件路径 this.fileReader = new FileReader(conf.get("wordsFile").toString()); } catch (FileNotFoundException e) { throw new RuntimeException("Error reading file [" + conf.get("wordFile") + "]"); } //初始化发射器 this.collector = collector; } /** * 这是Spout最主要的方法,在这里我们读取文本文件,并把它的每一行发射出去(给bolt) * 这个方法会不断被调用,为了降低它对CPU的消耗,当任务完成时让它sleep一下 * **/ @Override public void nextTuple() { if (completed) { try { Thread.sleep(1000); } catch (InterruptedException e) { // Do nothing } return; } String str; // Open the reader BufferedReader reader = new BufferedReader(fileReader); try { // Read all lines while ((str = reader.readLine()) != null) { /** * 发射每一行,Values是一个ArrayList的实现 */ this.collector.emit(new Values(str), str); } } catch (Exception e) { throw new RuntimeException("Error reading tuple", e); } finally { completed = true; } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("line")); } @Override public void close() { // TODO Auto-generated method stub } @Override public void activate() { // TODO Auto-generated method stub } @Override public void deactivate() { // TODO Auto-generated method stub } @Override public void ack(Object msgId) { System.out.println("OK:" + msgId); } @Override public void fail(Object msgId) { System.out.println("FAIL:" + msgId); } @Override public Map<String, Object> getComponentConfiguration() { // TODO Auto-generated method stub return null; } }
2.创建两个bolt来处理Spout发射出的数据
Spout已经成功读取文件并把每一行作为一个tuple(在Storm数据以tuple的形式传递)发射过来,我们这里需要创建两个bolt分别来负责解析每一行和对单词计数。
Bolt中最重要的是execute方法,每当一个tuple传过来时它便会被调用。
第一个bolt:WordNormalizer
package storm.demo.bolt; import java.util.ArrayList; import java.util.List; import java.util.Map; import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.IRichBolt; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; import backtype.storm.tuple.Values; public class WordNormalizer implements IRichBolt { private OutputCollector collector; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { this.collector = collector; } /**这是bolt中最重要的方法,每当接收到一个tuple时,此方法便被调用 * 这个方法的作用就是把文本文件中的每一行切分成一个个单词,并把这些单词发射出去(给下一个bolt处理) * **/ @Override public void execute(Tuple input) { String sentence = input.getString(0); String[] words = sentence.split(" "); for (String word : words) { word = word.trim(); if (!word.isEmpty()) { word = word.toLowerCase(); // Emit the word List a = new ArrayList(); a.add(input); collector.emit(a, new Values(word)); } } //确认成功处理一个tuple collector.ack(input); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word")); } @Override public void cleanup() { // TODO Auto-generated method stub } @Override public Map<String, Object> getComponentConfiguration() { // TODO Auto-generated method stub return null; } }
第二个bolt:WordCounter
package storm.demo.bolt; import java.util.HashMap; import java.util.Map; import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.IRichBolt; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Tuple; public class WordCounter implements IRichBolt { Integer id; String name; Map<String, Integer> counters; private OutputCollector collector; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { this.counters = new HashMap<String, Integer>(); this.collector = collector; this.name = context.getThisComponentId(); this.id = context.getThisTaskId(); } @Override public void execute(Tuple input) { String str = input.getString(0); if (!counters.containsKey(str)) { counters.put(str, 1); } else { Integer c = counters.get(str) + 1; counters.put(str, c); } // 确认成功处理一个tuple collector.ack(input); } /** * Topology执行完毕的清理工作,比如关闭连接、释放资源等操作都会写在这里 * 因为这只是个Demo,我们用它来打印我们的计数器 * */ @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()); } counters.clear(); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { // TODO Auto-generated method stub } @Override public Map<String, Object> getComponentConfiguration() { // TODO Auto-generated method stub return null; } }
3.在main函数中创建一个Topology
在这里我们要创建一个Topology和一个LocalCluster对象,还有一个Config对象做一些配置。
package storm.demo; import storm.demo.bolt.WordCounter; import storm.demo.bolt.WordNormalizer; import storm.demo.spout.WordReader; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.topology.TopologyBuilder; import backtype.storm.tuple.Fields; public class WordCountTopologyMain { public static void main(String[] args) throws InterruptedException { //定义一个Topology TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("word-reader",new WordReader()); builder.setBolt("word-normalizer", new WordNormalizer()) .shuffleGrouping("word-reader"); builder.setBolt("word-counter", new WordCounter(),2) .fieldsGrouping("word-normalizer", new Fields("word")); //配置 Config conf = new Config(); conf.put("wordsFile", "d:/text.txt"); conf.setDebug(false); //提交Topology conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1); //创建一个本地模式cluster LocalCluster cluster = new LocalCluster(); cluster.submitTopology("Getting-Started-Toplogie", conf, builder.createTopology()); Thread.sleep(1000); cluster.shutdown(); } }
运行这个函数我们即可看到后台打印出来的单词个数。
(ps:因为是Local模式,运行开始可能会打印很多错误log,这个先不用管)
以上所述就是小编给大家介绍的《Storm详解二、写第一个Storm应用》,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持!
猜你喜欢:- Zookeeper详解-应用程序(七)
- Go语言sync包的应用详解
- Go 语言 sync 包的应用详解
- 详解Spring Boot的应用限流
- 一看就懂,Python 日志模块详解及应用
- Golang性能测试工具PProf应用详解
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。