Storm 简介

https://github.com/nathanmarz/storm/wiki/Documentation

 

安装和配置

Storm的安装比较简单, 下载storm的release版本, 解压, 并且把bin/目录加到环境变量PATH里面去, 就ok了. 参考配置storm开发环境 
当然为了运行Storm, 需要装一些其他的依赖的包, 可以参考Twitter Storm 安装实战

Storm支持单机调试模式, 所以现在如果你已经有包含topology的jar包, 就可以直接运行单机模式来进行测试,

storm jar storm-starter-0.0.1-SNAPSHOT-standalone.jar storm.starter.WordCountTopology

当然实际使用时需要搭建Storm集群, 需要配置Zookeeper和Storm集群,并launch daemons(nimbus, supervisor, UI) , 可以参考Setting up a Storm cluster

对于storm命令, 其实可以理解为storm的客户端, 支持除了上面的jar以外的一系列操作, 参考Command line client

 

如何创建能在storm运行的Jar, Storm Starter

参考新建一个strom项目

https://github.com/nathanmarz/storm-starter

Storm Start给出很好的例子, 下载下来后用Leiningen(here)产生Jar 
Lein会根据Project.clj配置, 自动下载和添加依赖包

lein deps
lein compile

lein uberjar

生成Jar后, 就可以使用Storm命令测试

如果要新建项目只需要, 使用lein new创建项目, 然后将Project.clj参考Storm Starter的进行修改就ok了.

 

基本概念

参考, https://github.com/nathanmarz/storm/wiki/Concepts

       http://xumingming.sinaapp.com/138/twitter-storm%E5%85%A5%E9%97%A8/

 

Streams, 流

流作为storm的核心概念, 定义为无限的tuple序列.

什么是tuple? 
命名的value序列, 可以理解成Key/value序列, 每个value可以是任何类型, 动态类型不需要事先声明. 
Tuple在传输中需要序列化和反序列化, storm集成了普通类型的序列化模块, 用户可以自定义特殊类型的序列化逻辑 
A tuple is a named list of values, where each value can be any type. 
Tuples are dynamically typed -- the types of the fields do not need to be declared. 
Tuples have helper methods like getInteger and getString to get field values without having to cast the result. 
Storm needs to know how to serialize all the values in a tuple.

The stream is the core abstraction in Storm. A stream is an unbounded sequence of tuples that is processed and created in parallel in a distributed fashion. 
Streams are defined with a schema that names the fields in the stream's tuples. By default, tuples can contain integers, longs, shorts, bytes, strings, doubles, floats, booleans, and byte arrays. You can also define your own serializers so that custom types can be used natively within tuples.

Every stream is given an id when declared. Since single-stream spouts and bolts are so common, OutputFieldsDeclarer has convenience methods for declaring a single stream without specifying an id. In this case, the stream is given the default id of "default".

Resources:

Tuple: streams are composed of tuples

OutputFieldsDeclarer: used to declare streams and their schemas

Serialization: Information about Storm's dynamic typing of tuples and declaring custom serializations

ISerialization: custom serializers must implement this interface

CONFIG.TOPOLOGY_SERIALIZATIONS: custom serializers can be registered using this configuration

 

Spouts, 流的源头

Spout是Storm里面特有的名词, Stream的源头. 通常是从外部数据源读取tuples, 并emit到topology. 
Spout可以同时emit多个tuple stream, 通过OutputFieldsDeclarer中的declareStream method来定义 
Spout需要实现IRichSpout接口, 最重要的方法是nextTuple, storm会不断调用该接口从spout中取数据 
同时需要注意, Spout分为reliable or unreliable两种, 对于reliable, 还支持ack和fail方法, 具体参考"Reliability”

下面给出个nextTuple的例子, 从words里面随机选一个tuple作为输出

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));
}

A spout is a source of streams in a topology. 
Generally spouts will read tuples from an external source and emit them into the topology (e.g. a Kestrel queue or the Twitter API). 
Spouts can either be reliable or unreliable. A reliable spout is capable of replaying a tuple if it failed to be processed by Storm, whereas an unreliable spout forgets about the tuple as soon as it is emitted.

Spouts can emit more than one stream. To do so, declare multiple streams using the declareStream method of OutputFieldsDeclarer and specify the stream to emit to when using the emit method onSpoutOutputCollector.

The main method on spouts is nextTuple. nextTuple either emits a new tuple into the topology or simply returns if there are no new tuples to emit. It is imperative that nextTuple does not block for any spout implementation, because Storm calls all the spout methods on the same thread.

The other main methods on spouts are ack and fail. These are called when Storm detects that a tuple emitted from the spout either successfully completed through the topology or failed to be completed. ack and fail are only called for reliable spouts. See the Javadoc for more information.

Resources:

IRichSpout: this is the interface that spouts must implement.

Guaranteeing message processing

 

Bolts, 流的处理节点

对于Bolt, 用户可以定义任意的处理逻辑, 最重要的方法是execute, 输入为tuple, 输出为emit 0或多个tuples到OutputCollector. 

Bolt支持多个输入流和emit多个输出流, 输出流和spout一样, 通过OutputFieldsDeclarer中的declareStream method来定义; 对于输入流, 如果想subscribe上层节点的多个输出streaming, 需要显式的通过stream_id去订阅, 如果不明确指定stream_id, 默认会订阅default stream.

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"));
    }
}

All processing in topologies is done in bolts. Bolts can do anything from filtering, functions, aggregations, joins, talking to databases, and more.

Bolts can do simple stream transformations. Doing complex stream transformations often requires multiple steps and thus multiple bolts. For example, transforming a stream of tweets into a stream of trending images requires at least two steps: a bolt to do a rolling count of retweets for each image, and one or more bolts to stream out the top X images (you can do this particular stream transformation in a more scalable way with three bolts than with two).

Bolts can emit more than one stream. To do so, declare multiple streams using the declareStream method of OutputFieldsDeclarer and specify the stream to emit to when using the emit method onOutputCollector.

When you declare a bolt's input streams, you always subscribe to specific streams of another component. If you want to subscribe to all the streams of another component, you have to subscribe to each one individually. InputDeclarer has syntactic sugar for subscribing to streams declared on the default stream id. Saying declarer.shuffleGrouping("1") subscribes to the default stream on component "1" and is equivalent to declarer.shuffleGrouping("1", DEFAULT_STREAM_ID).

The main method in bolts is the execute method which takes in as input a new tuple. Bolts emit new tuples using the OutputCollector object. Bolts must call the ack method on the OutputCollector for every tuple they process so that Storm knows when tuples are completed (and can eventually determine that its safe to ack the original spout tuples). For the common case of processing an input tuple, emitting 0 or more tuples based on that tuple, and then acking the input tuple, Storm provides an IBasicBolt interface which does the acking automatically.

Its perfectly fine to launch new threads in bolts that do processing asynchronously. OutputCollector is thread-safe and can be called at any time.

Resources:

IRichBolt: this is general interface for bolts.

IBasicBolt: this is a convenience interface for defining bolts that do filtering or simple functions.

OutputCollector: bolts emit tuples to their output streams using an instance of this class

Guaranteeing message processing

 

Topologies, 拓扑

可以理解为类似MapReduce job 
根本区别, MR job执行完就结束, 而Topology会一直存在. 因为MR流动的是代码, 而Storm流动的数据. 
所以Storm不可能替代MR, 因为对于海量数据, 数据的流动是不合理的 
另一个区别, 我自己的想法, Topology对工作流有更好的支持, 而MR job往往只能完成一个map/reduce的过程, 而对于复杂的操作, 需要多个MR job才能完成. 
而Topology的定义更加灵活, 可以简单的使用一个topology支持比较复杂的工作流场景

Storm Topology是基于Thrift结构, 并且Nimbus是个Thrift server, 所以对于Topology可以用任何语言实现, 最终都是转化为Thrift结构

具体的Java版本的Topology的例子,

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"));

Topology有一个spout, 两个bolt. setSpout和setBolt的参数都是一样, 分别为id(在Topology中的唯一标识); 处理逻辑(对于Spout就是数据产生function); 并发线程数(task数) 
其中对于spout需要实现IRichSpout接口, 而bolt需要实现IRichBolt接口 
比较特别的是, setBolt方法会返回一个InputDeclarer对象, 并且该对象是用来定义Bolt输入的, 比如上面.shuffleGrouping(1), 用1(spout)的输出流作为输入

The logic for a realtime application is packaged into a Storm topology. A Storm topology is analogous to a MapReduce job. One key difference is that a MapReduce job eventually finishes, whereas a topology runs forever (or until you kill it, of course). A topology is a graph of spouts and bolts that are connected with stream groupings. These concepts are described below.

Since topology definitions are just Thrift structs, and Nimbus is a Thrift service, you can create and submit topologies using any programming language.

Resources:

TopologyBuilder: use this class to construct topologies in Java

Running topologies on a production cluster

Local mode: Read this to learn how to develop and test topologies in local mode.

 

Nimbus和Supervisor

在Storm的集群里面有两种节点: 
Nimbus节点, 主节点, 它的作用类似Hadoop里面的JobTracker. Nimbus is responsible for distributing code around the cluster, assigning tasks to machines, and monitoring for failures. 
Supervisor的节点, 工作节点, listens for work assigned to its machine and starts and stops worker processes as necessary based on what Nimbus has assigned to it. 
Each worker process executes a subset of a topology; a running topology consists of many worker processes spread across many machines.

Nimbus和Supervisor之间的所有协调工作都是通过一个Zookeeper集群来完成 
nimbus进程和supervisor都是快速失败（fail-fast)和无状态的, 所有的状态都存储在Zookeeper或本地磁盘上 
这也就意味着你可以用kill -9来杀死nimbus和supervisor进程, 然后再重启它们, 它们可以继续工作 
更重要的是, nimbus和supervisor的fail或restart不会影响worker的工作, 不象Hadoop, Job tracker的fail会导致job失败

 

Workers, Executor, Tasks

参考 Storm Topology的并发度

 

Stream groupings

如果从task的粒度来看一个运行的topology, 它应该如图, 所以需要策略决定blot和spout, 以及bolt之间的数据流向问题

Part of defining a topology is specifying for each bolt which streams it should receive as input. A stream grouping defines how that stream should be partitioned among the bolt's tasks.

There are seven built-in stream groupings in Storm, and you can implement a custom stream grouping by implementing the CustomStreamGrouping interface:

1. Shuffle grouping: Tuples are randomly distributed across the bolt's tasks in a way such that each bolt is guaranteed to get an equal number of tuples.
2. Fields grouping: The stream is partitioned by the fields specified in the grouping. For example, if the stream is grouped by the "user-id" field, tuples with the same "user-id" will always go to the same task, but tuples with different "user-id"'s may go to different tasks.
3. All grouping: The stream is replicated across all the bolt's tasks. Use this grouping with care.
4. Global grouping: The entire stream goes to a single one of the bolt's tasks. Specifically, it goes to the task with the lowest id.
5. None grouping: This grouping specifies that you don't care how the stream is grouped. Currently, none groupings are equivalent to shuffle groupings. Eventually though, Storm will push down bolts with none groupings to execute in the same thread as the bolt or spout they subscribe from (when possible).
6. Direct grouping: This is a special kind of grouping. A stream grouped this way means that the producer of the tuple decides which task of the consumer will receive this tuple. Direct groupings can only be declared on streams that have been declared as direct streams. Tuples emitted to a direct stream must be emitted using one of the emitDirect methods. A bolt can get the task ids of its consumers by either using the provided TopologyContext or by keeping track of the output of the emit method in OutputCollector (which returns the task ids that the tuple was sent to).
7. Local or shuffle grouping: If the target bolt has one or more tasks in the same worker process, tuples will be shuffled to just those in-process tasks. Otherwise, this acts like a normal shuffle grouping.

Resources:

TopologyBuilder: use this class to define topologies

InputDeclarer: this object is returned whenever setBolt is called on TopologyBuilder and is used for declaring a bolt's input streams and how those streams should be grouped

CoordinatedBolt: this bolt is useful for distributed RPC topologies and makes heavy use of direct streams and direct groupings

 

Reliability

Storm guarantees that every spout tuple will be fully processed by the topology. It does this by tracking the tree of tuples triggered by every spout tuple and determining when that tree of tuples has been successfully completed. Every topology has a "message timeout" associated with it. If Storm fails to detect that a spout tuple has been completed within that timeout, then it fails the tuple and replays it later.

To take advantage of Storm's reliability capabilities, you must tell Storm when new edges in a tuple tree are being created and tell Storm whenever you've finished processing an individual tuple. These are done using the OutputCollector object that bolts use to emit tuples. Anchoring is done in the emit method, and you declare that you're finished with a tuple using the ack method.

This is all explained in much more detail on Guaranteeing message processing (《storm如何保证消息不丢失》)