02、Storm课程入门到精通storm3-1

Storm深入学习Storm深入学习Storm数据模型(topology)Stormack和failStorm批处理StormTOPNStorm流程聚合StormDRPCStormexecutor、worker、task之间的关系和调优Storm异常解决Storm深入学习Storm数据模型(topology)

为了在storm做实时计算，必须创建topology。topology是计算图。topology中的每个节点包含一个处理逻辑，节点之间的链接表明了数据如何在节点之间被传输。

运行topology非常直接了当：首先将你的代码和依赖打包为一个jar，接着运行以下命令即可：

此命令运行类backtype.storm.MyTopology,参数为arg1和arg2。此类的主要功能定义了topology，并且提交到Nimbus。stormjar负责链接Nimbus上传jar。

因为topology定义是thrift结构，Nimbus是thrift服务，你可以使用任何语言创建，提交topology。Storm深入学习Storm数据模型(topology)

stream:

storm的核心是"stream"。stream是无边界的tuple序列。storm以分布、可靠的方式为转换一个stream到新的stream提供了基本组件。

storm为stream的转换提供的基本组件是spouts和bolts。spouts和bolt是你必须实现的接口，运行程序特定逻辑。

spout是stream的源，例如spout会读取kestrel队列的tuples，且作为一个stream发出。或spout会链接到twitterapi，作为tweetstream发送。

bolt消费多个inputstream，做一些处理，且可能发送新的stream。复杂的stream转换，如从tweetstream计算trendingtopics，需要多个步骤和多个bolts。bolts通过运行函数,过滤tuple，聚集，链接，与database交互可以实现任何操作。Storm深入学习Storm数据模型(topology) spouts和bolts的网络被包装成一个topology，这是提交给storm集群来执行的最顶层的抽象。一个topology是stream转换(每个节点是spout或bolt)的图。图中的边表示bolt订阅了哪些stream。当spout或bolt输出tuple到一个stream，它发送tuple到每个订阅此stream的bolt。Storm深入学习Storm数据模型(topology)

topology中nodes之间的链接表明tuple如何传递。例如spoutA->BoltB,spoutA->BoltC,BoltB->BoltC,spoutA发送的tuple会发送到BoltB和C，所有的BoltB的输出也会流向BoltC。

stormtopolgy中的每个节点的计算都是并行的。可以指定每个node的并行度，此时storm会在集群中产生指定并行度大小个线程来执行。

topology永不休止的运行，直到kill它。storm会自动重新分配失败的task。另外storm会保证没有数据丢失，即使机器停止，消息落下(遗失)。Storm深入学习Storm数据模型(topology)(Datamodel)数据模型 storm使用tuple作为它的数据模型。tuple是一命名的值的列表，tuple中的field可以是任何类型的对象。storm支持所有基本类型，string，字节数组。为了使用其他类型的对象，仅需要实现serializer接口。

在topology中的每个节点必须声明tuples的输出域。下例，bolt声明它输出两个tuples(double和triple两个域)。Storm深入学习Storm数据模型(topology)Storm深入学习Storm数据模型(topology)

declareoutputField函数声明了组件的["double","triple"]输出域。bolt的剩下部分会在后面解释。simpletopology

先看看简单的topology来探究更多的概念，看代码如何改进。Storm深入学习Storm数据模型(topology)

此topology包含一个spout和两个bolts。spout输处word，每个bolt追加"!!!"到输出。node排列成一行:spout输出到第一个bolt，此bolt输出到第二个bolt。

定义node的代码使用了setSpout和setBolt方法。这些方法接收用户自定义ID输入，一个包含处理逻辑的对象和node的并行度。包含处理逻辑的对象实现了IRichSpout和IRichBolt接口。最后一个参数:node的并行度，是可选的，它指定了在集群中多少个线程被创建来执行此组件，如果忽略，storm会为每个Node分配一个线程。 setBolt返回InputDeclarer对象，用来定义Bolt的输入。组件"exclaim1"声明它需要读取被组件"words"的所有tuple输出，使用shuffle分组。"shufflegroup"意味着tuple被随机从输入task分配到boltstask。这有很多方式在组件中对数据分组。Storm深入学习Storm数据模型(topology)

如果想要组件"exclaim2"读取组件"words"和"exclaim"的所有tuple输出，可以在"exclaim2"中这样定义：

输出声明可以链接起来，指定多个blot源。Storm深入学习Storm数据模型(topology)spouts和bolts实现 spouts负责输出新消息到topology。TestWordSpout输出从列表["nathan","mike","jackson","golda","bertels"]中随机输出word。Storm深入学习Storm数据模型(topology) ExclamationBolt追加"!!!"到它的输出。Storm深入学习Storm数据模型(topology)prepare方法给bolt个提供了OutputCollector，用来输出tuple。tuple可以在prepare，execute，cleanup等方法中任何时候输出，或在其他同步线程中。execute方法从bolt的输入接收tuple。ExcamationBolt从tuple中抓取第一个filed，并输出新的tuple。如果自己实现的bolt订阅了多个输入源，你可以通过Tuple.getSourceComponent方法来判断来自哪个组件。表面上，inputtuple作为第一个参数被传递输出，在最后一行调用ask()方法，这是storm的可靠API,来保证没有数据丢失。cleanup方法在Bolt被关闭，且清理任何打开的资源使用。不能保证此方法会在cluster上被调用，例如，如果机器的任务爆发性增长，不会调用此方法。cleanup方法更倾向于topology的本地模式。declareOutputFields方法声明ExclamationBolt输出getComponentConfiguration方法允许你配置此组件怎样运行的多个aspectStorm深入学习Storm数据模型(topology)cleanup和getComponentConfiguration方法在bolt实现中并不是经常需要。可以使用BaseClass来提供默认实现。Storm深入学习Storm数据模型(topology)本地模式运行Topology storm有两种模式：本地模式和分布式模式。在本地模式中，storm使用线程来模拟workernodes执行。本地模式用来测试和开发topologies。在分布式模式中，storm作为集群机器操作。当提交topology给master，你也提交所有必要的代码来运行topology。master利用你的分布式代码，分配workers运行你的topology。如果workers宕机了，master会在其他地方重新分配。Storm深入学习Storm数据模型(topology)Storm深入学习Storm数据模型(topology)

首先，代码通过创建的LocalCluster对象定义了in-processcluster。提交topology给虚拟的cluster，等同与提交topology给分布式cluster。它调用subumitTopology提交topology给LocalCluster，参数为要运行的topology的名称，topology的配置和topology自己。 topology的名字用来识别，可以在稍后kill它。configuration用来调节正在运行的topology的各方面。Storm深入学习Storm数据模型(topology)常见配置

有很多topology级的配置可以设。这里有关于所有配置的清单，以”TOPOLOGY”打头的配置是topology级别的配置，可以覆盖全局级别的配置。下面是一些比较常见的：1）Config.TOPOLOGY_WORKERS:

这个设置用多少个工作进程来执行这个topology。比如，如果你把它设置成25，那么集群里面一共会有25个java进程来执行这个topology的所有task。如果你的这个topology里面所有组件加起来一共有150的并行度，那么每个进程里面会有6个线程(150/25=6)。2）Config.TOPOLOGY_ACKERS:这个配置设置acker线程的数目。Ackers是Storm的可靠性API的一部分Storm深入学习Storm数据模型(topology)3）Config.TOPOLOGY_MAX_SPOUT_PENDING:

这个设置一个spouttask上面最多有多少个没有处理的tuple（没有ack/failed）回复，我们推荐你设置这个配置，以防止tuple队列爆掉。4）Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS:这个配置storm的tuple的超时时间

–超过这个时间的tuple被认为处理失败了。这个设置的默认设置是30秒，对于大多数的topology都已经足够了。5）Config.TOPOLOGY_SERIALIZATIONS:为了在你的tuple里面使用自定义类型，你可以用这个配置注册自定义serializer。6）TOPOLOGY_DEBUG,设置为true，记录被组件输出的每个消息。Storm深入学习Storm数据模型(topology)streamgrouping streamgrouping告诉topology在两个组件中怎样发送tuple。记住，spout和bolt在集群中并行以多个task执行。Storm深入学习Storm数据模型(topology)

当任务(BoltA输出一个tuple给BoltB)，此任务将tuple发送给哪个呢？streamgrouping告诉storm在多个task之间如何发送tuple。Storm深入学习Storm数据模型(topology)

splitSentence为每个单词输出一个tuple，wordCount在内存中保存一份从word到count的映射，每次wordcount接收一个word时，它更新状态，输出新word，count。

有多种streamgrouping。最简单的group，称为"shufflegrouping",它发送tuple给随机的task。更为有趣的"fieldgrouping"，它保证统一个word分配到统一个task。fieldsgrouping使得通过field的子集将stream分组。Fieldsgrouping是streamingjoin和streamingaggregation的实现基础。Storm深入学习Stormack和fail

为了保证数据能正确的被处理,对于spout产生的每一个tuple,storm都会进行跟踪,这里面涉及到ack/fail的处理,如果一个tuple处理成功,会调用spout的ack方法,如果失败,会调用fail方法.而在处理tuple的每一个bolt都会通过OutputCollector来告知storm,当前bolt处理是否成功.

IBasicBolt实现类不关心ack/fail,spout的ack/fail完全由后面的bolt的ack/fail来决定.其execute方法的BasicOutputCollector参数也没有提供ack/fail方法给你调用.相当于忽略了该bolt的ack/fail行为.所以IBasicBolt用来做filter或者简单的计算比较合适.

Storm深入学习Stormack和fail

参考BasicBoltExecutor代码里面的实现

Storm深入学习Stormack和fail

在IRichBolt实现类中,如果OutputCollector.emit(oldTuple,newTuple)这样调用来发射tuple(在storm中称之为anchoring),那么后面的bolt的ack/fail会影响spout的ack/fail,如果collector.emit(newTuple)这样来发射tuple(在storm称之为unanchoring),则相当于断开了后面bolt的ack/fail对sp