首页 > 代码库 > JStorm第一个程序WordCount详解
JStorm第一个程序WordCount详解
一、Strom基本知识（回顾）
1，首先明确Storm各个组件的作用，包括Nimbus,Supervisor,Spout,Bolt,Task,Worker,Tuple
- nimbus是整个storm任务的管理者，并不实际进行工作。负责在集群中分发代码，对节点分配任务，并监视主机故障。
- supervisor是实际进行工作的节点，负责监听工作节点上已经分配的主机作业，启动和停止Nimbus已经分配的工作进程。
- Worker是具体处理Spout/Bolt逻辑的进程，worker数量由拓扑中的conf.setNumWorkers来定义，storm会在每个Worker上均匀分配任务，一个Worker只能执行一个topology，但是可以执行其中的多个任务线程。
- 一个worker是一个进程，里面可以同时运行多个线程，这些线程就是task
参考：http://blog.csdn.net/cuihaolong/article/details/52652686（storm各个节点介绍和容错机制）
2，一个简单的storm程序的基本流程是：spout作为数据源（可以来自hdfs，hbase等，也可以自发产生数据，比如wordcount这个例子）传送给bolt，bolt对数据进行处理，传给其它bolt或者直接输出。他们之间传送的数据是Tuple,可以成为数据元组。
3，Storm运行模式：
- 本地模式(Local Mode)： 即Topology（相当于一个任务，后续会详细讲解） 运行在本地机器的单一JVM上，这个模式主要用来开发、调试。
- 远程模式(Remote Mode):在这个模式，我们把我们的Topology提交到集群，在这个模式中，Storm的所有组件都是线程安全的，因为它们都会运行在不同的Jvm或物理机器上，这个模式就是正式的生产模式。
二、WordCount详解
程序描述
- spout随机发送一个准备好的字符串数组里面的一个字符串（sentence）
- 第一层bolt，splitBolt负责对spout发过来的数据（sentence）进行split,分解成独立的单词，并按照一定的规则发往下一层bolt处理
- 第二层bolt，接受第一层bolt传过来的数据，并对各个单词进行数量计算
程序流程
- spout数据源
- bolt1进行split操作
- bolt2进行count操作
- Topolgy运行程序
0，WordCountTopology类：创建拓扑，运行程序
1 package act.chenkh.study.jstormPlay; 2 3 import backtype.storm.Config; 4 import backtype.storm.LocalCluster; 5 import backtype.storm.StormSubmitter; 6 import backtype.storm.topology.TopologyBuilder; 7 import backtype.storm.tuple.Fields; 8 9 public class WordCountTopology { 10 public static void main(String[] args) throws Exception { 11 /**第一步，设计一个Topolgy*/ 12 TopologyBuilder builder = new TopologyBuilder(); 13 /* 14 * 设置spout和bolt,完整参数为spout的id(即name),spout对象,并发数(此项没有默认null) 15 */ 16 //setSpout 17 builder.setSpout("sentence-spout",new RandomSentenceSpout(),1); 18 //setBolt:SplitBolt的grouping策略是上层随机分发，CountBolt的grouping策略是按照上层字段分发 19 //如果想要从多个Bolt获取数据，可以继续设置grouping 20 builder.setBolt("split-bolt", new SplitBolt(),1) 21 .shuffleGrouping("sentence-spout"); 22 builder.setBolt("count-bolt", new CountBolt(),1) 23 .fieldsGrouping("split-bolt", new Fields("word")); 24 /**第二步，进行基本配置*/ 25 Config conf = new Config(); 26 //作用和影响？？？？？？？？？？？ 27 conf.setDebug(true); 28 if (args != null && args.length > 0) { 29 conf.setNumWorkers(1); 30 StormSubmitter.submitTopology(args[0], conf, builder.createTopology()); 31 } 32 else { 33 /* 34 * run in local cluster, for test in eclipse. 35 */ 36 conf.setMaxTaskParallelism(3); 37 LocalCluster cluster = new LocalCluster(); 38 cluster.submitTopology("Getting-Started-Toplogie", conf, builder.createTopology()); 39 Thread.sleep(Integer.MAX_VALUE); 40 cluster.shutdown(); 41 } 42 } 43 }
1，RandomSentenceSpout类：产生数据
1 package act.chenkh.study.jstormPlay; 2 3 import backtype.storm.Config; 4 import backtype.storm.LocalCluster; 5 import backtype.storm.StormSubmitter; 6 import backtype.storm.topology.TopologyBuilder; 7 import backtype.storm.tuple.Fields; 8 9 public class WordCountTopology { 10 public static void main(String[] args) throws Exception { 11 /**第一步，设计一个Topolgy*/ 12 TopologyBuilder builder = new TopologyBuilder(); 13 /* 14 * 设置spout和bolt,完整参数为spout的id(即name),spout对象,并发数(此项没有默认null) 15 */ 16 //setSpout 17 builder.setSpout("sentence-spout",new RandomSentenceSpout(),1); 18 //setBolt:SplitBolt的grouping策略是上层随机分发，CountBolt的grouping策略是按照上层字段分发 19 //如果想要从多个Bolt获取数据，可以继续设置grouping 20 builder.setBolt("split-bolt", new SplitBolt(),1) 21 .shuffleGrouping("sentence-spout"); 22 builder.setBolt("count-bolt", new CountBolt(),1) 23 .fieldsGrouping("split-bolt", new Fields("word")); 24 /**第二步，进行基本配置*/ 25 Config conf = new Config(); 26 //作用和影响？？？？？？？？？？？ 27 conf.setDebug(true); 28 if (args != null && args.length > 0) { 29 conf.setNumWorkers(1); 30 StormSubmitter.submitTopology(args[0], conf, builder.createTopology()); 31 } 32 else { 33 /* 34 * run in local cluster, for test in eclipse. 35 */ 36 conf.setMaxTaskParallelism(3); 37 LocalCluster cluster = new LocalCluster(); 38 cluster.submitTopology("Getting-Started-Toplogie", conf, builder.createTopology()); 39 Thread.sleep(Integer.MAX_VALUE); 40 cluster.shutdown(); 41 } 42 } 43 }
2，SplitBolt类：接收上层tuple,进行split,分发给下一层
1 package act.chenkh.study.jstormPlay; 2 3 import java.util.HashMap; 4 import java.util.Map; 5 6 import org.apache.log4j.Logger; 7 8 import com.alibaba.jstorm.callback.AsyncLoopThread; 9 import com.alibaba.jstorm.callback.RunnableCallback; 10 11 import backtype.storm.task.TopologyContext; 12 import backtype.storm.topology.BasicOutputCollector; 13 import backtype.storm.topology.OutputFieldsDeclarer; 14 import backtype.storm.topology.base.BaseBasicBolt; 15 import backtype.storm.tuple.Fields; 16 import backtype.storm.tuple.Tuple; 17 import clojure.inspector__init; 18 19 public class CountBolt extends BaseBasicBolt { 20 Integer id; 21 String name; 22 Map<String, Integer> counters; 23 String component; 24 private static final Logger LOG = Logger.getLogger(CountBolt.class); 25 private AsyncLoopThread statThread; 26 /** 27 * On create 28 */ 29 @Override 30 public void prepare(Map stormConf, TopologyContext context) { 31 this.counters = new HashMap<String, Integer>(); 32 this.name = context.getThisComponentId(); 33 this.id = context.getThisTaskId(); 34 this.statThread = new AsyncLoopThread(new statRunnable()); 35 36 LOG.info(stormConf.get("abc")+"!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"); 37 component = context.getThisComponentId(); 38 } 39 40 public void declareOutputFields(OutputFieldsDeclarer declarer) { 41 declarer.declare(new Fields("word","count")); 42 // declarer.declareStream("coord-"+"word-counter", new Fields("epoch","ebagNum")); 43 // LOG.info("set stream coord-"+component); 44 } 45 46 //接收消息之后被调用的方法 47 public void execute(Tuple input, BasicOutputCollector collector) { 48 // String str = input.getString(0); 49 String str = input.getStringByField("word"); 50 if(!counters.containsKey(str)){ 51 counters.put(str, 1); 52 }else{ 53 Integer c = counters.get(str) + 1; 54 counters.put(str, c); 55 } 56 } 57 class statRunnable extends RunnableCallback { 58 59 @Override 60 public void run() { 61 while(true){ 62 try { 63 Thread.sleep(10000); 64 } catch (InterruptedException e) { 65 66 } 67 LOG.info("\n-- Word Counter ["+name+"-"+id+"] --"); 68 for(Map.Entry<String, Integer> entry : counters.entrySet()){ 69 LOG.info(entry.getKey()+": "+entry.getValue()); 70 } 71 LOG.info(""); 72 } 73 74 } 75 } 76 77 }
3，CountBolt类：接收上层tuple,进行count,展示输出
1 package act.chenkh.study.jstormPlay; 2 3 import java.util.HashMap; 4 import java.util.Map; 5 6 import org.apache.log4j.Logger; 7 8 import com.alibaba.jstorm.callback.AsyncLoopThread; 9 import com.alibaba.jstorm.callback.RunnableCallback; 10 11 import backtype.storm.task.TopologyContext; 12 import backtype.storm.topology.BasicOutputCollector; 13 import backtype.storm.topology.OutputFieldsDeclarer; 14 import backtype.storm.topology.base.BaseBasicBolt; 15 import backtype.storm.tuple.Fields; 16 import backtype.storm.tuple.Tuple; 17 import clojure.inspector__init; 18 19 public class CountBolt extends BaseBasicBolt { 20 Integer id; 21 String name; 22 Map<String, Integer> counters; 23 String component; 24 private static final Logger LOG = Logger.getLogger(CountBolt.class); 25 private AsyncLoopThread statThread; 26 /** 27 * On create 28 */ 29 @Override 30 public void prepare(Map stormConf, TopologyContext context) { 31 this.counters = new HashMap<String, Integer>(); 32 this.name = context.getThisComponentId(); 33 this.id = context.getThisTaskId(); 34 this.statThread = new AsyncLoopThread(new statRunnable()); 35 36 LOG.info(stormConf.get("abc")+"!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"); 37 component = context.getThisComponentId(); 38 } 39 40 public void declareOutputFields(OutputFieldsDeclarer declarer) { 41 declarer.declare(new Fields("word","count")); 42 // declarer.declareStream("coord-"+"word-counter", new Fields("epoch","ebagNum")); 43 // LOG.info("set stream coord-"+component); 44 } 45 46 //接收消息之后被调用的方法 47 public void execute(Tuple input, BasicOutputCollector collector) { 48 // String str = input.getString(0); 49 String str = input.getStringByField("word"); 50 if(!counters.containsKey(str)){ 51 counters.put(str, 1); 52 }else{ 53 Integer c = counters.get(str) + 1; 54 counters.put(str, c); 55 } 56 } 57 class statRunnable extends RunnableCallback { 58 59 @Override 60 public void run() { 61 while(true){ 62 try { 63 Thread.sleep(10000); 64 } catch (InterruptedException e) { 65 66 } 67 LOG.info("\n-- Word Counter ["+name+"-"+id+"] --"); 68 for(Map.Entry<String, Integer> entry : counters.entrySet()){ 69 LOG.info(entry.getKey()+": "+entry.getValue()); 70 } 71 LOG.info(""); 72 } 73 74 } 75 } 76 77 }
参考：http://fireinwind.iteye.com/blog/2153699（第一个Storm应用）
三、Grouping的几种方式
四、Bolt的声明周期
1、在定义Topology实例过程中，定义好Spout实例和Bolt实例
2、在提交Topology实例给Nimbus的过程中，会调用TopologyBuilder实例的createTopology()方法，以获取定义的Topology实例。在运行createTopology()方法的过程中，会去调用Spout和Bolt实例上的declareOutputFields()方法和getComponentConfiguration()方法，declareOutputFields()方法配置Spout和Bolt实例的输出，getComponentConfiguration()方法输出特定于Spout和Bolt实例的配置参数值对。Storm会将以上过程中得到的实例，输出配置和配置参数值对等数据序列化，然后传递给Nimbus。
3、在Worker Node上运行的thread，从Nimbus上复制序列化后得到的字节码文件，从中反序列化得到Spout和Bolt实例，实例的输出配置和实例的配置参数值对等数据，在thread中Spout和Bolt实例的declareOutputFields()和getComponentConfiguration()不会再运行。
4、在thread中，反序列化得到一个Bolt实例后，它会先运行Bolt实例的prepare()方法，在这个方法调用中，需要传入一个OutputCollector实例，后面使用该OutputCollector实例输出Tuple
5、接下来在该thread中按照配置数量建立task集合，然后在每个task中就会循环调用thread所持有Bolt实例的execute()方法
6、在关闭一个thread时，thread所持有的Bolt实例会调用cleanup()方法
不过如果是强制关闭，这个cleanup()方法有可能不会被调用到
五、Stream里面的Tuple
1，Stream是storm里面的关键抽象。一个stream是一个没有边界的tuple序列。
storm提供一些原语来分布式地、可靠地把一个stream传输进一个新的stream。比如： 你可以把一个tweets流传输到热门话题的流。
storm提供的最基本的处理stream的原语是spout和bolt。你可以实现Spout和Bolt对应的接口以处理你的应用的逻辑。
spout的流的源头。比如一个spout可能从Kestrel队列里面读取消息并且把这些消息发射成一个流。又比如一个spout可以调用twitter的一个api并且把返回的tweets发射成一个流。
通常Spout会从外部数据源（队列、数据库等）读取数据，然后封装成Tuple形式，之后发送到Stream中。Spout是一个主动的角色，在接口内部有个nextTuple函数，Storm框架会不停的调用该函数。
bolt可以接收任意多个输入stream， 作一些处理， 有些bolt可能还会发射一些新的stream。一些复杂的流转换， 比如从一些tweet里面计算出热门话题， 需要多个步骤， 从而也就需要多个bolt。 Bolt可以做任何事情: 运行函数， 过滤tuple, 做一些聚合， 做一些合并以及访问数据库等等。
Bolt处理输入的Stream，并产生新的输出Stream。Bolt可以执行过滤、函数操作、Join、操作数据库等任何操作。Bolt是一个被动的 角色，其接口中有一个execute(Tuple input)方法，在接收到消息之后会调用此函数，用户可以在此方法中执行自己的处理逻辑。
spout和bolt所组成一个网络会被打包成topology， topology是storm里面最高一级的抽象（类似 Job）， 你可以把topology提交给storm的集群来运行。topology的结构在Topology那一段已经说过了，这里就不再赘述了。
参考：http://www.cnblogs.com/wuxiang/p/5629138.html（Storm入门原理介绍）
2，Tuple: 消息传递的基本单位。
在spout发送的时候，函数原型
1 public List<Integer> emit(List<Object> tuple, Object messageId) { 2 return emit(Utils.DEFAULT_STREAM_ID, tuple, messageId); 3 }
这里的tuple, 实际上是List<Object> 对象，返回的是 List<Integer> 是要发送的tast的IdsList
在bolt接收的时候, 变成一个Tuple对象, 结构应该也是一个list， List<Field1, value1, Field2, value2..>这样的一个结构， FieldList ValueList, 我们根据对应的fieldname就可以取出对应的getIntegerByField方法
JStorm第一个程序WordCount详解
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。