《Storm入门介绍》由会员分享，可在线阅读，更多相关《Storm入门介绍（9页珍藏版）》请在金锄头文库上搜索。
1、. . Storm介绍周龙鹏一、数据处理过程Storm的术语解释Storm的术语包括Stream、Spout、Bolt、Task、Worker、Stream Grouping和Topology。Stream是被处理的数据。Spout是数据源。Bolt处理数据。Task是运行于Spout或Bolt中的线程。Worker是运行这些线程的进程。Stream Grouping规定了Bolt接收什么东西作为输入数据。数据可以随机分配（术语为Shuffle），或者根据字段值分配（术语为Fields），或者 广播（术语为All），或者总是发给一个Task（术语为Global），也可以不关心该数据（术语为No
2、ne），或者由自定义逻辑来决定（术语为 Direct）。Topology是由Stream Grouping连接起来的Spout和Bolt节点网络。在Storm Concepts页面里对这些术语有更详细的描述。（1）Topologies 用于封装一个实时计算应用程序的逻辑，类似于Hadoop的MapReduce Job（2）Stream 消息流，是一个没有边界的tuple序列，这些tuples会被以一种分布式的方式并行地创建和处理（3）Spouts 消息源，是消息生产者，他会从一个外部源读取数据并向topology里面面发出消息：tuple（4）Bolts 消息处理者，所有的消息处理逻辑被封装在
3、bolts里面，处理输入的数据流并产生输出的新数据流，可执行过滤，聚合，查询数据库等操作（5）Task 每一个Spout和Bolt会被当作很多task在整个集群里面执行，每一个task对应到一个线程.二、storm集群的组件（topologies）Storm集群非常类似Hadoop集群。Hadoop上运行的是MapReduce jobs，而Storm运行的是topologies。Jobs和topologies本身是不同的，其中一个最大的不同就是，Mapreduce job最终会完毕，而topology则会持续的处理消息（直到你杀掉它）。Storm集群有2种节点：master（主节点）和work
4、er（工作节点）。 master节点运行一个守护进程，叫Nimbus，类似Hadoop中的JobTracker。Nimbus负责在集群中分发代码，分配任务，以与故障检测。 每个worker节点运行一个守护进程，叫Supervisor。Supervisor监听分配到该服务器的任务，开始和完毕工作进程。每个worker进程执行topology的一个子集；一个运行中的topology由许多分布在多台机器上的worker进程组成。Nimbus和Supervisors之间是通过Zookeeper协调。此外，Nimbus和Supervisor是能快速失败（fail-fast）和无状态的（stateless
5、）；所有的状态都保存在Zookeeper或者在本地磁盘中。这意味这你可以kill 9杀掉Nimbus或者Supervisors，随后它们会自动恢复好像什么也没发生过。这项设计使得Storm集群变得非常稳定健壮。三、Topologies在Storm中进行实时计算，你可以创建所谓的topologies。一个topology是一个计算图（a graph of computation）。topology的每个节点包括处理逻辑，节点之间数据如何传输的连接。运行一个topology非常简单。将代码和相关依赖打包成一个简单jar包，运行如下命令：storm jar all-my-code.jar backt
6、ype.storm.MyTopology arg1 arg2将执行这个类：backtype.storm.MyTopology，参数arg1和arg2。这个类的最主要功能定义了topology，并将其提交给Nimbus。storm jar 部分负责连接Nimbus和上传jar。因为topology定义是Thrift结构，Nimbus是一个Thrift service，所以你可以使用任何编程语言创建和提交topologies。上面的例子是最简单的方式，使用JVM-based的语言。四、StreamsStorm中最核心的抽象就是stream。stream是一个无边界的tuples序列。Storm提供
7、了基本流转换的分布式的可靠的方法。例如，你可以将tweets流转换成一个trending topics流。基本的原始Storm提供了spouts和bolts的流转换。Spouts和bolts提供了接口，实现你的应用逻辑。spout是流源头。例如，一个spout可以是从Kestrel队列中读取tuples并以流形式发射（emit）出。或者一个spout可以连接Twitter API，发出一个tweets流。一个bolt使用任意数量的输入流，做些处理，可能再发射出新的流。复杂的流转换，例如从tweets流中计算出trending topics流，需要多个步骤，所以需要多个bolts。bolts可以
8、做任何事情，包括运行函数、过滤tuples、流的聚合、流连接、数据库交互等等。spouts和bolts的网络封装到topology，后者是你提交给storm集群运行的最上层的抽象。一个topology是一个由spout或bolt节点组成的流变换的图，图的边指示哪个bolts订阅给哪个stream。当一个spout或bolt发射一个tuple给stream，它将发送这个tuple到订阅了这个stream的每一个bolt。在topology中，节点之间的连接指示这tuples如何传递。例如，如果Spout A和Bolt B之间存在连接，Spout A和Bolt C之间存在连接，Bolt B和Bol
9、t C之间存在连接，那么每时刻Spout A发送出一个tuple，它将发给Bolt B和Bolt C，所有Bolt B的输出tuple将都会发给Bolt C。每个Storm topology中，在各个节点中都是并行执行。在topology中，你可以指定每个节点的并行数，然后Storm将会开启相应数量的线程来运行。每个topology会永久运行，直到你kill它。Storm会自动再分配失败的任务。此外，Storm会保证不会有数据丢失，即使是宕机消息丢失。五、Data ModelStorm使用tuples作为它的数据模型。每个tuple是值的名字序列（a named list of values）
10、，且tuple中域（field）可以是任何类型对象。Storm支持所有的原始类型，strings、byte arrays，作为tuple field values。如果要使用新类型对象，需要为这个类型实现一个序列化（a serializer）。每个节点都必须声明它发送出的tuple的输出域。例如，这个bolt声明了它发送2个tuple，对应的域类型是double和tripleThedeclareOutputFieldsfunction declares the output fieldsdouble, triplefor the component. The rest of the bolt
11、will be explained in the upcoming sections.六、A simple topology来看看一个简单topology实例，深入探索相关的概念和编码。下面是ExclamationTopology的定义：这个topology包括1个spout和2个bolts。这个spout发送words，每个bolt将输入字符串附加上“！”。各节点安排在一条线：spout发给第一个bolt，后者在发给第二个bolt。例如spout发送“bob” and “john”，经过两个bolt，将会发送出“bob!” and “john!”。This code defines the
12、nodes using thesetSpoutandsetBoltmethods. These methods take as input a user-specified id, an object containing the processing logic, and the amount of parallelism you want for the node. In this example, the spout is given id words and the bolts are given ids exclaim1 and exclaim2.这个对象包含两个接口的处理逻辑的实现
13、：spout的一个接口 IRichSpout ，bolt的一个接口 IRichBolt。最后一个参数，你希望节点上并行数，是可选项。它声明了这个集群中由多少个线程来运行这个组件。如果省略这个参数，Storm缺省下只分配一个线程给一个节点。setBolt返回一个 InputDeclarer 对象，用于声明Bolt的输入。这个例子中，“exclaim1”定义了它需要读取所有来自“words”组件以shuffle grouping方式发出的tuples，“exclaim2”定义了需要读取所有来自“exclaim1”组件以shuffle grouping方式发出的tuples。shuffle grou
14、ping意思是将tuples从输入任务中随机分配给bolts任务。关于组件之间分组数据的方式将后面grouping章节。如果你想让组件exclaim2同时读取words组件和exclaim1组件两者的tuples，你可以这个定义exclaim2，输入定义可以指定多个来源，形成链式：As you can see, input declarations can be chained to specify multiple sources for the Bolt.进一步看看这个topology中spout和bolt的实现。Spouts负责产生新的消息给这个topology。TestWordSpou
15、t 是从一个列表中每隔100ms随机选择单词生成一个tuple。在 TestWordSpout 的nextTuple()实现如下：As you can see, the implementation is very straightforward.ExclamationBoltappends the string ! to its input. Lets take a look at the full implementation forExclamationBolt:其中prepare方法提供给bolt一个OutputCollector，用于发送出一个tuple。Tuple可以被任何时间发送，在prepare、execute、cleanup方法，甚至是其他线程中异步方式。这里的prepare实现的很简单，将OutputCollector存成一个实例变量，在后面的execute方法中使用。其中execute方法是从一个bolt输入中接收一个tuple，ExclaimationBolt提取tuple中第一个域，附加上“！”成一个新字符串，发送成一个新的tuple。如果你实现一个bolt读取多个输入源，你可以使用Tuple#getSourceComponent方法找到Tuple来自哪个组件。execute中执行了一些
