storm学习-storm入门

超好资料：

英文：https://github.com/xetorthio/getting-started-with-storm/blob/master/ch03Topologies.asc

中文：http://ifeve.com/getting-started-with-storm-3/

下面具体讲下：storm的几种groupping 策略的例子

Storm Grouping

shuffleGrouping

将流分组定义为混排。这种混排分组意味着来自Spout的输入将混排，或随机分发给此Bolt中的任务。shuffle grouping对各个task的tuple分配的比较均匀。
fieldsGrouping

这种grouping机制保证相同field值的tuple会去同一个task，这对于WordCount来说非常关键，如果同一个单词不去同一个task，那么统计出来的单词次数就不对了。
All grouping

广播发送，对于每一个tuple将会复制到每一个bolt中处理。
Global grouping

Stream中的所有的tuple都会发送给同一个bolt任务处理，所有的tuple将会发送给拥有最小task_id的bolt任务处理。
None grouping

不关注并行处理负载均衡策略时使用该方式，目前等同于shuffle grouping,另外storm将会把bolt任务和他的上游提供数据的任务安排在同一个线程下。
Direct grouping

由tuple的发射单元直接决定tuple将发射给那个bolt，一般情况下是由接收tuple的bolt决定接收哪个bolt发射的Tuple。这是一种比较特别的分组方法，用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。只有被声明为Direct Stream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发射。消息处理者可以通过TopologyContext来获取处理它的消息的taskid (OutputCollector.emit方法也会返回taskid)

In this chapter, we’ll see how to pass tuples between the different components of a Storm topology, and how to deploy a topology into a running Storm cluster

Stream Grouping

One of the most important things that we need to do when designing a topology is to define how data is exchanged between components (how streams are consumed by the bolts). A Stream Grouping specifies which stream(s) are consumed by each bolt and how the stream will be consumed.

Tip	A node can emit more than one stream of data. A stream grouping allows us to choose which stream to receive.

The stream grouping is set when the topology is defined, as we saw in chapter 2, Getting Started:

....
    builder.setBolt("word-normalizer", new WordNormalizer())
        .shuffleGrouping("word-reader");
....

Here a bolt is set on the topology builder, and then a source is set using the shuffle stream grouping. A stream grouping normally takes the source component id as a parameter, and optionally other parameters as well, depending on the kind of stream grouping.

Tip	There can be more than one source per `InputDeclarer`, and each source can be grouped with a different stream grouping.

Shuffle Grouping

Shuffle Grouping is the most commonly used grouping. It takes a single parameter (the source component), and sends each tuple, emitted by the source, to a randomly chosen bolt warranting that each consumer will receive the same number of tuples .

The shuffle grouping is useful for doing atomic operations. For example, a math operation. However if the operation can’t be randomically distributed, such as the example in chapter 2 where we needed to count words, we should considerate the use of other grouping.

Fields Grouping

Fields Grouping allows us to control how tuples are sent to bolts, based on one or more fields of the tuple. It guarantees that a given set of values, for a combination of fields, is always sent to the same bolt. Coming back to the word count example, if we group the stream by the word field, the word-normalizer bolt will always send tuples with a given word to the same instance of the word-counter bolt.

....
builder.setBolt("word-counter", new WordCounter(),2)
    .fieldsGrouping("word-normalizer", new Fields("word"));
....

Tip	All fields set in the fields grouping must exist in the sources’s field declaration.

All Grouping

All Grouping sends a single copy of each tuple to all instances of the receiving bolt. This kind of grouping is used to sendsignals to bolts, for example if we need to refresh a cache we can send a refresh cache signal to all bolts. In the word-count example, we could use an all grouping to add the ability to clear the counter cache (see Topologies Example)

    public void execute(Tuple input) {
        String str = null;
        try{
            if(input.getSourceStreamId().equals("signals")){
                str = input.getStringByField("action");
                if("refreshCache".equals(str))
                    counters.clear();
            }
        }catch (IllegalArgumentException e) {
            //Do nothing
        }
        ....
    }

We’ve added an if to check the stream source. Storm give us the posibility to declare named streams (if we don’t send a tuple to a named stream the stream is "default") it’s an excelent way to identify the source of the tuples like this case where we want to identify the signals

In the topology definition, we add a second stream to the word-counter bolt that sends each tuple from the signals-spout stream to all instances of the bolt.

builder.setBolt("word-counter", new WordCounter(),2)
            .fieldsGrouping("word-normalizer", new Fields("word"))
            .allGrouping("signals-spout","signals");

The implementation of signals-spout can be found at git repository.

Custom Grouping

We can create our own custom stream grouping by implementing the backtype.storm.grouping.CustomStreamGroupinginterface. This gives us the power to decide which bolt(s) will receive each tuple.

Let’s modify the word count example, to group tuples so that all words that start with the same letter will be received by the same bolt.

public class ModuleGrouping implements CustomStreamGrouping, Serializable{

    int numTasks = 0;

    @Override
    public List<Integer> chooseTasks(List<Object> values) {
        List<Integer> boltIds = new ArrayList();
        if(values.size()>0){
            String str = values.get(0).toString();
            if(str.isEmpty())
                boltIds.add(0);
            else
                boltIds.add(str.charAt(0) % numTasks);
        }
        return boltIds;
    }

    @Override
    public void prepare(TopologyContext context, Fields outFields,
            List<Integer> targetTasks) {
        numTasks = targetTasks.size();
    }
}

Here we can see a simple implementation of CustomStreamGrouping, where we use the amount of tasks to take the modulus of the integer value of the first character of the word, thus selecting which bolt will receive the tuple.

To use this grouping in our example we should change the word-normalizer grouping by the next:

       builder.setBolt("word-normalizer", new WordNormalizer())
            .customGrouping("word-reader", new ModuleGrouping());

Direct Grouping

This is a special grouping where the source decides which component will receive the tuple. Similarly to the previous example, the source will decide which bolt receives the tuple based on the first letter of the word. To use direct grouping, in the WordNormalizer bolt we use the emitDirect method instead of emit.

    public void execute(Tuple input) {
        ....
        for(String word : words){
            if(!word.isEmpty()){
                ....
                collector.emitDirect(getWordCountIndex(word),new Values(word));
            }
        }
        // Acknowledge the tuple
        collector.ack(input);
    }

    public Integer getWordCountIndex(String word) {
        word = word.trim().toUpperCase();
        if(word.isEmpty())
            return 0;
        else
            return word.charAt(0) % numCounterTasks;
    }

We work out the number of target tasks in the prepare method:

    public void prepare(Map stormConf, TopologyContext context,
            OutputCollector collector) {
        this.collector = collector;
        this.numCounterTasks = context.getComponentTasks("word-counter");
    }

And in the topology definition, we specify that the stream will be grouped directly:

    builder.setBolt("word-counter", new WordCounter(),2)
            .directGrouping("word-normalizer");

Global grouping

Global Grouping sends tuples generated by all instances of the source to a single target instance (specifically, the task with lowest id).

None grouping

At the time of writing (storm version 0.7.1), using this grouping is the same as using Shuffle Grouping. In other words, when using this grouping, we don’t care how streams are grouped