Storm（实时Hadoop）指南
Storm的主要特点如下：
- 简单的编程模型。类似于MapReduce降低了并行批处理复杂性，Storm降低了进行实时处理的复杂性。
- 可以使用各种编程语言。你可以在Storm之上使用各种编程语言。默认支持Clojure、Java、Ruby和Python。要增加对其他语言的支持，只需实现一个简单的Storm通信协议即可。
- 容错性。Storm会管理工作进程和节点的故障。
- 水平扩展。计算是在多个线程、进程和服务器之间并行进行的。
- 可靠的消息处理。Storm保证每个消息至少能得到一次完整处理。任务失败时，它会负责从消息源重试消息。
- 快速。系统的设计保证了消息能得到快速的处理，使用?MQ作为其底层消息队列。
- 本地模式。Storm有一个“本地模式”，可以在处理过程中完全模拟Storm集群。这让你可以快速进行开发和单元测试。
Storm集群由一个主节点和多个工作节点组成。主节点运行了一个名为“Nimbus”的守护进程，用于分配代码、布置任务及故障检测。每个工作节 点都运行了一个名为“Supervisor”的守护进程，用于监听工作，开始并终止工作进程。Nimbus和Supervisor都能快速失败，而且是无 状态的，这样一来它们就变得十分健壮，两者的协调工作是由Apache ZooKeeper来完成的。
Storm的术语包括Stream、Spout、Bolt、Task、Worker、Stream Grouping和Topology。Stream是被处理的数据。Sprout是数据源。Bolt处理数据。Task是运行于Spout或Bolt中的线程。Worker是运行这些线程的进程。Stream Grouping规定了Bolt接收什么东西作为输入数据。数据可以随机分配（术语为Shuffle），或者根据字段值分配（术语为Fields），或者 广播（术语为All），或者总是发给一个Task（术语为Global），也可以不关心该数据（术语为None），或者由自定义逻辑来决定（术语为 Direct）。Topology是由Stream Grouping连接起来的Spout和Bolt节点网络。在Storm Concepts页面里对这些术语有更详细的描述。
可以和Storm相提并论的系统有Esper、Streambase、HStreaming和Yahoo S4。其中和Storm最接近的就是S4。两者最大的区别在于Storm会保证消息得到处理。这些系统中有的拥有内建数据存储层，这是Storm所没有的，如果需要持久化，可以使用一个类似于Cassandra或Riak这样的外部数据库。
入门的最佳途径是阅读GitHub上的官方《Storm Tutorial》。 其中讨论了多种Storm概念和抽象，提供了范例代码以便你可以运行一个Storm Topology。开发过程中，可以用本地模式来运行Storm，这样就能在本地开发，在进程中测试Topology。一切就绪后，以远程模式运行 Storm，提交用于在集群中运行的Topology。Maven用户可以使用clojars.org提供的Storm依赖，地址是 http://clojars.org/repo。
在这个教程里面我们将学习如何创建Topologies, 并且把topologies部署到storm的集群里面去。Java将是我们主要的示范语言， 个别例子会使用python以演示storm的多语言特性。
准备工作
这个教程使用storm-starter项目里面的例子。我推荐你们下载这个项目的代码并且跟着教程一起做。先读一下：配置storm开发环境和新建一个strom项目这两篇文章把你的机器设置好。
一个Storm集群的基本组件
storm的集群表面上看和hadoop的集群非常像。但是在Hadoop上面你运行的是MapReduce的Job, 而在Storm上面你运行的是Topology。它们是非常不一样的 — 一个关键的区别是： 一个MapReduce Job最终会结束， 而一个Topology运永远运行（除非你显式的杀掉他）。
在Storm的集群里面有两种节点： 控制节点(master node)和工作节点(worker node)。控制节点上面运行一个后台程序： Nimbus， 它的作用类似Hadoop里面的JobTracker。Nimbus负责在集群里面分布代码，分配工作给机器， 并且监控状态。
每一个工作节点上面运行一个叫做Supervisor的节点。Supervisor会监听分配给它那台机器的工作，根据需要 启动/关闭工作进程。每一个工作进程执行一个Topology的一个子集；一个运行的Topology由运行在很多机器上的很多工作进程组成。
storm topology结构
Nimbus和Supervisor之间的所有协调工作都是通过一个Zookeeper集群来完成。并且，nimbus进程和supervisor 经常都是快速失败（fail-fast)和无状态的。所有的状态要么在Zookeeper里面， 要么在本地磁盘上。这也就意味着你可以用kill -9来杀死nimbus和supervisor进程， 然后再重启它们，它们可以继续工作， 就好像什么都没有发生过似的。这个设计使得storm不可思议的稳定。
Topologies
为了在storm上面做实时计算， 你要去建立一些topologies。一个topology就是一个计算节点所组成的图。Topology里面的每个处理节点都包含处理逻辑， 而节点之间的连接则表示数据流动的方向。
运行一个Topology是很简单的。首先，把你所有的代码已经所依赖的jar打进一个jar包。然后运行类似下面的这个命令。
strom jar all-your-code.jar backtype.storm.MyTopology arg1 arg2 |
这个命令会运行主类: backtype.strom.MyTopology, 参数是arg1, arg2。这个类的main函数定义这个topology并且把它提交给Nimbus。storm jar负责连接到nimbus并且上传jar文件。
因为topology的定义其实就是一个Thrift结构并且nimbus就是一个Thrift服务， 有可以用任何语言创建并且提交topology。上面的方面是用JVM
-based语言提交的最简单的方法, 看一下文章: Running topologies on a production cluster去看看怎么启动和停止topologies。
Stream
Stream是storm里面的关键抽象。一个stream是一个没有边界的tuple序列。storm提供一些语言来以分布式地，可靠地把一个stream传输进一个新的stream。比如： 你可以把一个tweets流传输到热门话题的流。
storm提供的最基本的处理stream的原语是spout和bolt。Spout和bolt有接口可以实现以让你实现你的应用的逻辑。
spout的流的源头。比如一个spout可能从Kestrel队列里面读取消息并且把这些消息发射成一个流。又比如一个spout可以调用twitter的一个api并且把返回的tweets发射成一个流。
bolt可以接收任意多个输入stream， 作一些处理， 有些bolt可能还会发射一些新的stream。一些复杂的流转换， 比如从一些tweet里面计算出热门话题， 需要多个步骤， 从而也就需要多个bolt。 Bolt可以做任何事情: 运行函数， 过滤tuple, 做一些聚合， 做一些合并以及访问数据库等等。
spout和bolt所组成一个网络会被打包成topology， topology是storm里面最高一级的抽象， 你可以把topology提交给storm的集群来运行。topology的结构在Topology那一段已经说过了，这里就不再赘述了。
topology结构
topology里面的每一个节点都是并行运行的。 在你的topology里面， 你可以指定每个节点的并行度， storm则会在集群里面分配那么多线程来同时计算。
一个topology会一直运行直到你杀死它。storm自动重新分配一些运行失败的任务， 并且storm保证你不会有数据丢失， 即使在一些机器意外停机并且消息被丢掉的情况下。
数据模型(Data Model)
storm使用tuple来作为它的数据模型。每个tuple是一堆值，每个值有一个名字，并且每个值可以是任何类型， 在我的理解里面一个tuple可以看作一个没有方法的java对象。总体来看，storm支持所有的基本类型，字符串以及字节数组作为tuple的值类 型。你也可以使用你自己定义的类型来作为值类型， 只要你实现对应的序列化器(serializer)。
topology里面的每个节点必须定义它要发射的tuple的每个字段。 比如下面这个bolt定义它锁发射的tuple包含两个字段，类型分别是: doble和triple。
public class DoubleAndTripleBolt implements IRichBolt {
private OutputCollectorBase _collector; @Override public void prepare(Map conf, TopologyContext context, OutputCollectorBase collector) {
_collector = collector; } @Override public void execute(Tuple input) {
int val = input.getInteger(0); _collector.emit(input, new Values(val*2, val*3)); _collector.ack(input); } @Override public void cleanup() {
} @Override public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("double", "triple"));
} } |
declareOutputFields方法定义要输出的字段 ： [“double”, “triple”]。这个bolt的其它部分我们接下来会解释。
一个简单的Topology
让我们来看一个简单的topology的例子， 我们看一下storm-starter里面的ExclamationTopology:
TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(1, new TestWordSpout(), 10); builder.setBolt(2, new ExclamationBolt(), 3) .shuffleGrouping(1); builder.setBolt(3, new ExclamationBolt(), 2) .shuffleGrouping(2); |
这个Topology包含一个Spout和两个Bolt。Spout发射单词， 每个bolt在每个单词后面加个”!!!”。这三个节点被排成一条线: spout发射单词给第一个bolt， 第一个bolt然后把处理好的单词发射给第二个bolt。如果spout发射的单词是[“bob”]和[“john”], 那么第二个bolt会发射[“bolt!!!!!!”]和[“john!!!!!!”]出来。
我们使用setSpout和setBolt来定义Topology里面的节点。这些方法接收我们指定的一个id， 一个包含处理逻辑的对象(spout或者bolt), 以及你所需要的并行度。
这个包含处理的对象如果是spout那么要实现IRichSpout的接口， 如果是bolt，那么就要实现IRichBolt接口.
最后一个指定并行度的参数是可选的。它表示集群里面需要多少个thread来一起执行这个节点。如果你忽略它那么storm会分配一个线程来执行这个节点。
setBolt方法返回一个InputDeclarer对象， 这个对象是用来定义Bol的输入。 这里第一个Bolt声明它要读取spout所发射的所有的tuple — 使用shuffle grouping。而第二个bolt声明它读取第一个bolt所发射的tuple。shuffle grouping表示所有的tuple会被随机的分发给bolt的所有task。给task分发tuple的策略有很多种，后面会介绍。
如果你想第二个bolt读取spout和第一个bolt所发射的所有的tuple， 那么你应该这样定义第二个bolt:
builder.setBolt(3, new ExclamationBolt(), 5) .shuffleGrouping(1) .shuffleGrouping(2); |
让我们深入地看一下这个topology厘米的spout和bolt是怎么实现的。Spout负责发射新的tuple到这个topology里面 来。TestWordSpout从[“nathan”, “mike”, “jackson”, “golda”, “bertels”]里面随机选择一个单词发射出来。TestWordSpout里面的nextTuple()方法是这样定义的：
public void nextTuple() {
Utils.sleep(100); final String[] words = new String[] {"nathan", "mike", "jackson", "golda", "bertels"};
final Random rand = new Random(); final String word = words[rand.nextInt(words.length)]; _collector.emit(new Values(word)); } |
可以看到，实现很简单。
ExclamationBolt把”!!!”拼接到输入tuple后面。我们来看下ExclamationBolt的完整实现。
public static class ExclamationBolt implements IRichBolt {
OutputCollector _collector; public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
_collector = collector; } public void execute(Tuple tuple) {
_collector.emit(tuple, new Values(tuple.getString(0) + "!!!")); _collector.ack(tuple); } public void cleanup() {
} public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
} } |
prepare方法提供给bolt一个Outputcollector用来发射tuple。Bolt可以在任何时候发射tuple — 在prepare, execute或者cleanup方法里面, 或者甚至在另一个线程里面异步发射。这里prepare方法只是简单地把OutputCollector作为一个类字段保存下来给后面execute方法 使用。
execute方法从bolt的一个输入接收tuple(一个bolt可能有多个输入源). ExclamationBolt获取tuple的第一个字段，加上”!!!”之后再发射出去。如果一个bolt有多个输入源，你可以通过调用 Tuple#getSourceComponent方法来知道它是来自哪个输入源的。
execute方法里面还有其它一些事情值得一提： 输入tuple被作为emit方法的第一个参数，并且输入tuple在最后一行被ack。这些呢都是Storm可靠性API的一部分，后面会解释。
cleanup方法在bolt被关闭的时候调用， 它应该清理所有被打开的资源。但是集群不保证这个方法一定会被执行。比如执行task的机器down掉了，那么根本就没有办法来调用那个方法。 cleanup设计的时候是被用来在local mode的时候才被调用(也就是说在一个进程里面模拟整个storm集群), 并且你想在关闭一些topology的时候避免资源泄漏。
最后，declareOutputFields定义一个叫做”word”的字段的tuple。
以local mode运行ExclamationTopology
让我们看看怎么以local mode运行ExclamationToplogy。
storm的运行有两种模式: 本地模式和分布式模式. 在本地模式中， storm用一个进程里面的线程来模拟所有的spout和bolt. 本地模式对开发和测试来说比较有用。 你运行storm-starter里面的topology的时候它们就是以本地模式运行的， 你可以看到topology里面的每一个组件在发射什么消息。
在分布式模式下， storm由一堆机器组成。当你提交topology给master的时候， 你同时也把topology的代码提交了。master负责分发你的代码并且负责给你的topolgoy分配工作进程。如果一个工作进程挂掉了， master节点会把认为重新分配到其它节点。关于如何在一个集群上面运行topology， 你可以看看Running topologies on a production cluster文章。
下面是以本地模式运行ExclamationTopology的代码:
Config conf = new Config(); conf.setDebug(true);conf.setNumWorkers(2); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("test", conf, builder.createTopology());
Utils.sleep(10000); cluster.killTopology("test");
cluster.shutdown(); |
首先， 这个代码定义通过定义一个LocalCluster对象来定义一个进程内的集群。提交topology给这个虚拟的集群和提交topology给分布式集 群是一样的。通过调用submitTopology方法来提交topology， 它接受三个参数：要运行的topology的名字，一个配置对象以及要运行的topology本身。
topology的名字是用来唯一区别一个topology的，这样你然后可以用这个名字来杀死这个topology的。前面已经说过了， 你必须显式的杀掉一个topology， 否则它会一直运行。
Conf对象可以配置很多东西， 下面两个是最常见的：
1. TOPOLOGY_WORKERS(setNumWorkers) 定义你希望集群分配多少个工作进程给你来执行这个topology. topology里面的每个组件会被需要线程来执行。每个组件到底用多少个线程是通过setBolt和setSpout来指定的。这些线程都运行在工作进 程里面. 每一个工作进程包含一些节点的一些工作线程。比如， 如果你指定300个线程，60个进程， 那么每个工作进程里面要执行6个线程， 而这6个线程可能属于不同的组件(Spout, Bolt)。你可以通过调整每个组件的并行度以及这些线程所在的进程数量来调整topology的性能。
2.TOPOLOGY_DEBUG(setDebug), 当它被设置成true的话， storm会记录下每个组件所发射的每条消息。这在本地环境调试topology很有用， 但是在线上这么做的话会影响性能的。
感兴趣的话可以去看看Conf对象的Javadoc去看看topology的所有配置。
可以看看创建一个新storm项目去看看怎么配置开发环境以使你能够以本地模式运行topology.
Stream grouping
Stream grouping告诉topology如何在两个组件之间发送tuple。 要记住， spouts和bolts以很多task的形式在topology里面同步执行。如何从task粒度来看一个运行的topology， 它应该是这样的:
从task角度来看topology
当Bolt A的一个task要发送一个tuple给Bolt B， 它应该发送给Bolt B的哪个task呢？
stream grouping专门回答这种问题的。在我们深入研究不同的stream grouping之前， 让我们看一下storm-starter里面的另外一个topology。WordCountTopology读取一些句子， 输出句子里面每个单词出现的次数.
TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(1, new RandomSentenceSpout(), 5); builder.setBolt(2, new SplitSentence(), 8) .shuffleGrouping(1); builder.setBolt(3, new WordCount(), 12) .fieldsGrouping(2, new Fields("word"));
|
SplitSentence对于句子里面的每个单词发射一个新的tuple, WordCount在内存里面维护一个单词->次数的mapping， WordCount每收到一个单词， 它就更新内存里面的统计状态。
有好几种不同的stream grouping:
最简单的grouping是shuffle grouping, 它随机发给任何一个task。上面例子里面RandomSentenceSpout和SplitSentence之间用的就是shuffle grouping, shuffle grouping对各个task的tuple分配的比较均匀。
一种更有趣的grouping是fields grouping, SplitSentence和WordCount之间使用的就是fields grouping, 这种grouping机制保证相同field值的tuple会去同一个task， 这对于WordCount来说非常关键，如果同一个单词不去同一个task， 那么统计出来的单词次数就不对了。
fields grouping是stream合并，stream聚合以及很多其它场景的基础。在背后呢， fields grouping使用的一致性哈希来分配tuple的。
还有一些其它类型的stream grouping. 你可以在Concepts一章里更详细的了解。
使用别的语言来定义Bolt
Bolt可以使用任何语言来定义。用其它语言定义的bolt会被当作subprocess来执行， storm使用JSON消息通过stdin/stdout来和这些subprocess通信。这个通信协议是一个只有100行的库， storm团队给这些库开发了对应的Ruby, Python和Fancy版本。
下面是WordCountTopology里面的SplitSentence的定义:
public static class SplitSentence extends ShellBolt implements IRichBolt {
public SplitSentence() {
super("python", "splitsentence.py");
} public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
} } |
SplitSentence继承自ShellBolt并且声明这个Bolt用python来运行，并且参数是: splitsentence.py。下面是splitsentence.py的定义:
import storm class SplitSentenceBolt(storm.BasicBolt): def process(self, tup): words = tup.values[0].split(" ")
for word in words: storm.emit([word]) SplitSentenceBolt().run() |
更多有关用其它语言定义Spout和Bolt的信息， 以及用其它语言来创建topology的 信息可以参见: Using non-JVm languages with Storm.
可靠的消息处理
在这个教程的前面，我们tuple发射的一些方面。这些方面就是storm的可靠性API： storm如果保证spout发出的每一个tuple都被完整处理。看看《storm如何保证消息不丢失》以更深入了解storm的可靠性API.
结论
这个入门教程比较广泛的介绍了从开发，测试和部署一个topology. 文档的其它部分会深入介绍使用storm的各个方面。
storm保证从spout发出的每个tuple都会被完全处理。这篇文章介绍storm是怎么做到这个保证的，以及我们使用者怎么做才能充分利用storm的可靠性特点。
一个tuple被”完全处理”是什么意思？
就如同蝴蝶效应一样，从spout发射的一个tuple可以引起其它成千上万个tuple因它而产生， 想想那个计算一篇文章中每个单词出现次数的topology.
TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(1, new KestrelSpout("kestrel.backtype.com",
22133, "sentence_queue", new StringScheme())); builder.setBolt(2, new SplitSentence(), 10) .shuffleGrouping(1); builder.setBolt(3, new WordCount(), 20) .fieldsGrouping(2, new Fields("word"));
|
这个topology从一个Kestrel队列读取句子， 把每个句子分割成一个个单词， 然后发射这一个个单词： 一个源tuple（一个句子）引起后面很多tuple的产生(一个个单词）， 这个消息流大概是这样的：
统计单词出现次数的tuple树
在storm里面一个tuple被完全处理的意思是： 这个tuple以及由这个tuple所导致的所有的tuple都被成功处理。而一个tuple会被认为处理失败了如果这个消息在timeout所指定的时间内没有成功处理。 而这个timetout可以通过Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS来指定。
如果一个消息处理成功了或者失败了会发生什么？
FYI。 下面这个是spout要实现的接口:
public interface ISpout extends Serializable {
void open(Map conf, TopologyContext context, SpoutOutputCollector collector); void close(); void nextTuple(); void ack(Object msgId); void fail(Object msgId); } |
首先storm通过调用spout的nextTuple方法来获取下一个tuple, Spout通过open方法参数里面提供的SpoutOutputCollector来发射新tuple到它的其中一个输出消息流, 发射tuple的时候spout会提供一个message-id, 后面我们通过这个tuple-id来追踪这个tuple。举例来说， KestrelSpout从kestrel队列里面读取一个消息，并且把kestrel提供的消息id作为message-id, 看例子：
_collector.emit(new Values("field1", "field2", 3) , msgId);
|
接下来， 这个发射的tuple被传送到消息处理者bolt那里， storm会跟踪由此所产生的这课tuple树。如果storm检测到一个tuple被完全处理了， 那么storm会以最开始的那个message-id作为参数去调用消息源的ack方法；反之storm会调用spout的fail方法。值得注意的一点 是， storm调用ack或者fail的task始终是产生这个tuple的那个task。所以如果一个spout被分成很多个task来执行， 消息执行的成功失败与否始终会通知最开始发出tuple的那个task。
我们再以KestrelSpout为例来看看spout需要做些什么才能保证“一个消息始终被完全处理”, 当KestrelSpout从Kestrel里面读出一条消息， 首先它“打开”这条消息， 这意味着这条消息还在kestrel队列里面， 不过这条消息会被标示成“处理中”直到ack或者fail被调用。处于“处理中“状态的消息不会被发给其他消息处理者了；并且如果这个spout“断线” 了， 那么所有处于“处理中”状态的消息会被重新标示成“等待处理”.
Storm的可靠性API
作为storm的使用者，有两件事情要做以更好的利用storm的可靠性特征。 首先，在你生成一个新的tuple的时候要通知storm; 其次，完成处理一个tuple之后要通知storm。 这样storm就可以检测整个tuple树有没有完成处理，并且通知源spout处理结果。storm提供了一些简洁的api来做这些事情。
由一个tuple产生一个新的tuple称为： anchoring。你发射一个新tuple的同时也就完成了一次anchoring。看下面这个例子： 这个bolt把一个包含一个句子的tuple分割成每个单词一个tuple。
public class SplitSentence implements IRichBolt {
OutputCollector _collector; public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
_collector = collector; } public void execute(Tuple tuple) {
String sentence = tuple.getString(0); for(String word: sentence.split(" ")) {
_collector.emit(tuple, new Values(word)); } _collector.ack(tuple); } public void cleanup() {
} public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
} } |
看一下这个execute方法， emit的第一个参数是输入tuple， 第二个参数则是输出tuple， 这其实就是通过输入tuple anchoring了一个新的输出tuple。因为这个“单词tuple”被anchoring在“句子tuple”一起， 如果其中一个单词处理出错，那么这整个句子会被重新处理。作为对比， 我们看看如果通过下面这行代码来发射一个新的tuple的话会有什么结果。
_collector.emit(new Values(word)); |
用这种方法发射会导致新发射的这个tuple脱离原来的tuple树(unanchoring), 如果这个tuple处理失败了， 整个句子不会被重新处理。到底要anchoring还是要 unanchoring则完全取决于你的业务需求。
一个输出tuple可以被anchoring到多个输入tuple。这种方式在stream合并或者stream聚合的时候很有用。一个多入口tuple处理失败的话，那么它对应的所有输入tuple都要重新执行。看看下面演示怎么指定多个输入tuple:
List <Tuple> anchors = new ArrayList<Tuple>(); anchors.add(tuple1); anchors.add(tuple2); _collector.emit(anchors, new Values(1, 2, 3)); |
多入口tuple把这个新tuple加到了多个tuple树里面去了。
我们通过anchoring来构造这个tuple树，最后一件要做的事情是在你处理完当个tuple的时候告诉storm, 通过OutputCollector类的ack和fail方法来做，如果你回过头来看看SplitSentence的例子， 你可以看到“句子tuple”在所有“单词tuple”被发出之后调用了ack。
你可以调用OutputCollector 的fail方法去立即将从消息源头发出的那个tuple标记为fail， 比如你查询了数据库，发现一个错误，你可以马上fail那个输入tuple， 这样可以让这个tuple被快速的重新处理， 因为你不需要等那个timeout时间来让它自动fail。
每个你处理的tuple， 必须被ack或者fail。因为storm追踪每个tuple要占用内存。所以如果你不ack/fail每一个tuple， 那么最终你会看到OutOfMemory错误。
大多数Bolt遵循这样的规律：读取一个tuple；发射一些新的tuple；在execute的结束的时候ack这个tuple。这些Bolt往 往是一些过滤器或者简单函数。Storm为这类规律封装了一个BasicBolt类。如果用BasicBolt来做， 上面那个SplitSentence可以改写成这样：
public class SplitSentence implements IBasicBolt {
public void prepare(Map conf, TopologyContext context) {
} public void execute(Tuple tuple, BasicOutputCollector collector) {
String sentence = tuple.getString(0); for(String word: sentence.split(" ")) {
collector.emit(new Values(word)); } } public void cleanup() {
} public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
} } |
这个实现比之前的实现简单多了， 但是功能上是一样的。发送到BasicOutputCollector的tuple会自动和输入tuple相关联，而在execute方法结束的时候那个输入tuple会被自动ack的。
作为对比，处理聚合和合并的bolt往往要处理一大堆的tuple之后才能被ack， 而这类tuple通常都是多输入的tuple， 所以这个已经不是IBasicBolt可以罩得住的了。
storm是怎么实现高效率的可靠性的？
storm里面有一类特殊的task称为：acker， 他们负责跟踪spout发出的每一个tuple的tuple树。当acker发现一个tuple树已经处理完成了。它会发送一个消息给产生这个tuple 的那个task。你可以通过Config.TOPOLOGY_ACKERS来设置一个topology里面的acker的数量， 默认值是一。 如果你的topology里面的tuple比较多的话， 那么把acker的数量设置多一点，效率会高一点。
理解storm的可靠性的最好的方法是来看看tuple和tuple树的生命周期， 当一个tuple被创建， 不管是spout还是bolt创建的， 它会被赋予一个64位的id，而acker就是利用这个id去跟踪所有的tuple的。每个tuple知道它的祖宗的id(从spout发出来的那个 tuple的id), 每当你新发射一个tuple， 它的祖宗id都会传给这个新的tuple。所以当一个tuple被ack的时候，所以当一个tuple被处理完的时候，它会发一个消息给acker，告诉 它这个tuple树发生了怎么样的变化。具体来说就是：它告诉acker： 我呢已经完成了， 我有这些儿子tuple, 你跟踪一下他们吧。下面这个图演示了C被ack了之后，这个tuple树所发生的变化。
tuple ack示例
关于storm怎么跟踪tuple还有一些细节， 前面已经提到过了， 你可以自己设定你的topology里面有多少个acker。而这又给我们带来一个问题， 当一个tuple需要ack的时候，它到底选择哪个tuple来ack呢？
storm使用一致性哈希来把一个spout-tuple-id对应到acker， 因为每一个tuple知道它所有的祖宗的tuple-id， 所以它自然可以算出要通知哪个acker来ack。
storm的另一个细节是acker是怎么知道每一个spout tuple应该交给哪个task来处理。当一个spout发射一个新的tuple， 它会简单的发一个消息给一个合适的acker，并且告诉acker它自己的id(taskid)， 这样storm就有了taskid-tupleid的对应关系。 当acker发现一个树完成处理了， 它知道给哪个task发送成功的消息。
acker task并不显式的跟踪tuple树。对于那些有成千上万个节点的tuple树，把这么多的tuple信息都跟踪起来会耗费太多的内存。相反， acker用了一种不同的方式， 使得对于每个spout tuple所需要的内存量是恒定的（20 bytes) . 这个跟踪算法是storm如何工作的关键，并且也是它的主要突破。
一个acker task存储了一个spout-tuple-id到一对值的一个mapping。这个对子的第一个值是创建这个tuple的taskid， 这个是用来在完成处理tuple的时候发送消息用的。 第二个值是一个64位的数字称作：”ack val”, ack val是整个tuple树的状态的一个表示，不管这棵树多大。它只是简单地把这棵树上的所有创建的tupleid/ack的tuple一起异或 (XOR)。
当一个acker task 发现一个 ack val变成0了， 它知道这棵树已经处理完成了。 因为tupleid是随机的64位数字， 所以， ack val碰巧变成0(而不是因为所有创建的tuple都完成了)的几率极小。算一下就知道了， 就算每秒发生10000个ack， 那么需要50000000万年才可能碰到一个错误。而且就算碰到了一个错误， 也只有在这个tuple失败的时候才会造成数据丢失。
既然你已经理解了storm的可靠性算法， 让我们一起过一遍所有可能的失败场景，并看看storm在每种情况下是怎么避免数据丢失的。
1. 一个tuple没有ack： 因为处理他的task挂掉了。storm的超时机制在超时之后会把这个tuple标记为失败，从而可以重新处理。
2. Acker挂掉了： 这种情况下由这个acker所跟踪的所有spout tuple都会超时，也就会被重新处理。
3. Spout挂掉了： 在这种情况下给spout发送消息的消息源负责重新发送这些消息。比如Kestrel和RabbitMQ在一个客户端断开之后会把所有”处理中“的消息放回队列。
就像你看到的那样， storm的可靠性机制是完全分布式的， 可伸缩的并且是高度容错的。
调整可靠性 (Tuning Reliability)
acker task是非常轻量级的， 所以一个topology里面不需要很多acker。你可以通过Strom UI(id: -1)来跟踪它的性能。 如果它的吞吐量看起来不正常，那么你就需要多加点acker了。
如果可靠性对你来说不是那么重要 — 你不太在意在一些失败的情况下损失一些数据， 那么你可以通过不跟踪这些tuple树来获取更好的性能。不去跟踪消息的话会使得系统里面的消息数量减少一半， 因为对于每一个tuple都要发送一个ack消息。并且它需要更少的id来保存下游的tuple， 减少带宽占用。
有三种方法可以去掉可靠性。第一是把Config.TOPOLOGY_ACKERS 设置成 0. 在这种情况下， storm会在spout发射一个tuple之后马上调用spout的ack方法。也就是说这个tuple树不会被跟踪。
第二个方法是在tuple层面去掉可靠性。 你可以在发射tuple的时候不指定messageid来达到不跟粽某个特定的spout tuple的目的。
最后一个方法是如果你对于一个tuple树里面的某一部分到底成不成功不是很关心，那么可以在发射这些tuple的时候unanchor它们。 这样这些tuple就不在tuple树里面， 也就不会被跟踪了。