Storm集群和Hadoop集群表面上看很类似。但是Hadoop上运行的是MapReduce jobs，而在Storm上运行的是拓扑（topology），这两者之间是非常不一样的。Topology的定义是一个Thrift结构，并且Nimbus就是一个Thrift服务， 你可以提交由任何语言创建的topology。

 

 

二、构建Topology

1. 实现的目标：

我们将设计一个topology，来实现对一个句子里面的单词出现的频率进行统计。这是一个简单的例子，目的是让大家对于topology快速上手，有一个初步的理解。

2. 设计Topology结构：

在开始开发Storm项目的第一步，就是要设计topology。确定好你的数据处理逻辑，我们今天将的这个简单的例子，topology也非常简单。整个topology如下：

 

整个topology分为三个部分：

• KestrelSpout:数据源，负责发送sentence
• Splitsentence:负责将sentence切分
• Wordcount:负责对单词的频率进行累加

3. 设计数据流

这个topology从kestrel queue读取句子,并把句子划分成单词,然后汇总每个单词出现的次数,一个tuple负责读取句子,每一个tuple分别对应计算每一个单词出现的次数,大概样子如下所示：

 

4. 代码实现：

1) 构建maven环境：

为了开发storm topology, 你需要把storm相关的jar包添加到classpath里面去： 要么手动添加所有相关的jar包， 要么使用maven来管理所有的依赖。storm的jar包发布在Clojars(一个maven库), 如果你使用maven的话，把下面的配置添加在你项目的pom.xml里面。

1. <repository> 
2.     <id>clojars.org</id> 
3.     <url>http://clojars.org/repo</url> 
4. </repository> 
5. <dependency> 
6.      <groupId>storm</groupId> 
7.     <artifactId>storm</artifactId> 
8.      <version>0.5.3</version> 
9.      <scope>test</scope> 
10. </dependency> 

2) 定义topology：

1. TopologyBuilder builder = new TopologyBuilder(); 
2. builder.setSpout(1, new KestrelSpout(“kestrel.backtype.com”,22133,                                                                                   ”sentence_queue”,                                                                                 new StringScheme())); 
3. builder.setBolt(2, new SplitSentence(), 10) 
4. .shuffleGrouping(1); 
5. builder.setBolt(3, new WordCount(), 20) 
6. .fieldsGrouping(2, new Fields(“word”)); 

这种topology的spout从句子队列中读取句子，在kestrel.backtype.com位于一个Kestrel的服务器端口22133。

Spout用setSpout方法插入一个独特的id到topology。 Topology中的每个节点必须给予一个id，id是由其他bolts用于订阅该节点的输出流。 KestrelSpout在topology中id为1。

setBolt是用于在Topology中插入bolts。 在topology中定义的第一个bolts 是切割句子的bolts。 这个bolts 将句子流转成成单词流。

让我们看看SplitSentence实施：

1. public class SplitSentence implements IBasicBolt{ 
2.         public void prepare(Map conf, TopologyContext context) { 
3.          } 
4.        public void execute(Tuple tuple, BasicOutputCollector collector) { 
5.               String sentence = tuple.getString(0); 
6.                for(String word: sentence.split(“ ”)) { 
7.                         collector.emit(new Values(word)); 
8.                   } 
9.              } 
10.          public void cleanup() { 
11.         } 
12.         public void declareOutputFields(OutputFieldsDeclarer declarer) { 
13.                 declarer.declare(new Fields(“word”)); 
14.              } 
15.  } 

关键的方法是 execute方法。 正如你可以看到，它将句子拆分成单词，并发出每个单词作为一个新的元组。 另一个重要的方法是declareOutputFields，其中宣布bolts输出元组的架构。 在这里宣布，它发出一个域为word的元组。

setBolt的最后一个参数是你想为bolts的并行量。 SplitSentence bolts 是10个并发，这将导致在storm集群中有十个线程并行执行。 你所要做的的是增加bolts的并行量在遇到topology的瓶颈时。

setBolt方法返回一个对象，用来定义bolts的输入。 例如，SplitSentence螺栓订阅组件“1”使用随机分组的输出流。 “1”是指已经定义KestrelSpout。 我将解释在某一时刻的随机分组的一部分。 到目前为止，最要紧的是，SplitSentence bolts会消耗KestrelSpout发出的每一个元组。

下面在让我们看看wordcount的实现：

1. public class WordCount implements IBasicBolt { 
2.         private Map<String, Integer> _counts = new HashMap<String, Integer>(); 
3.         public void prepare(Map conf, TopologyContext context) { 
4.         } 
5.        public void execute(Tuple tuple, BasicOutputCollector collector) { 
6.               String word = tuple.getString(0); 
7.               int count; 
8.               if(_counts.containsKey(word)) { 
9.                      count = _counts.get(word); 
10.               } else { 
11.                     count = 0; 
12. } 
13.               count++; 
14.               _counts.put(word, count); 
15.               collector.emit(new Values(word, count)); 
16.        } 
17.        public void cleanup() { 
18.        } 
19.        public void declareOutputFields(OutputFieldsDeclarer declarer) { 
20.               declarer.declare(new Fields(“word”, “count”)); 
21.        } 
22. } 

SplitSentence对于句子里面的每个单词发射一个新的tuple, WordCount在内存里面维护一个单词->次数的mapping， WordCount每收到一个单词， 它就更新内存里面的统计状态。

5. 运行Topology

storm的运行有两种模式: 本地模式和分布式模式.

1) 本地模式：

storm用一个进程里面的线程来模拟所有的spout和bolt. 本地模式对开发和测试来说比较有用。 你运行storm-starter里面的topology的时候它们就是以本地模式运行的， 你可以看到topology里面的每一个组件在发射什么消息。

2) 分布式模式：

storm由一堆机器组成。当你提交topology给master的时候， 你同时也把topology的代码提交了。master负责分发你的代码并且负责给你的topolgoy分配工作进程。如果一个工作进程挂掉了， master节点会把认为重新分配到其它节点。

3) 下面是以本地模式运行的代码：

1. Config conf = new Config(); 
2. conf.setDebug(true); 
3. conf.setNumWorkers(2); 
4. LocalCluster cluster = new LocalCluster(); 
5. cluster.submitTopology(“test”, conf, builder.createTopology()); 
6. Utils.sleep(10000); 
7. cluster.killTopology(“test”); 
8. cluster.shutdown(); 

首先， 这个代码定义通过定义一个LocalCluster对象来定义一个进程内的集群。提交topology给这个虚拟的集群和提交topology给分布式集群是一样的。通过调用submitTopology方法来提交topology， 它接受三个参数：要运行的topology的名字，一个配置对象以及要运行的topology本身。

topology的名字是用来唯一区别一个topology的，这样你然后可以用这个名字来杀死这个topology的。前面已经说过了， 你必须显式的杀掉一个topology， 否则它会一直运行。

Conf对象可以配置很多东西， 下面两个是最常见的：

TOPOLOGY_WORKERS(setNumWorkers) 定义你希望集群分配多少个工作进程给你来执行这个topology. topology里面的每个组件会被需要线程来执行。每个组件到底用多少个线程是通过setBolt和setSpout来指定的。这些线程都运行在工作进程里面. 每一个工作进程包含一些节点的一些工作线程。比如， 如果你指定300个线程，60个进程， 那么每个工作进程里面要执行6个线程， 而这6个线程可能属于不同的组件(Spout, Bolt)。你可以通过调整每个组件的并行度以及这些线程所在的进程数量来调整topology的性能。

TOPOLOGY_DEBUG(setDebug), 当它被设置成true的话， storm会记录下每个组件所发射的每条消息。这在本地环境调试topology很有用， 但是在线上这么做的话会影响性能的。

结论：

本章从storm的基本对象的定义，到广泛的介绍了storm的开发环境，从一个简单的例子讲解了topology的构建和定义。希望大家可以从本章的内容对storm有一个基本的理解和概念，并且已经可以构建一个简单的topology！！