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.

样例
此样例由Storm的官方提供，通过下面这个例子可以对Shuffle grouping有更直观的认识

public class ExclamationTopology {

public static class ExclamationBolt extends BaseRichBolt {
    OutputCollector _collector;

    @Override
    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
        _collector = collector;
    }

    @Override
    public void execute(Tuple tuple) {
        System.out.println(tuple.getString(0) + " is from task " + tuple.getSourceTask() + " of Spout/Bolt:" + tuple.getSourceComponent());

        _collector.emit(tuple, new Values(tuple.getString(0) + "!!!"));
        _collector.ack(tuple);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));
    }

}

public static void main(String[] args) throws Exception {
    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout("word", new TestWordSpout(), 10);
    builder.setBolt("exclaim1", new ExclamationBolt(), 3).shuffleGrouping("word");
    builder.setBolt("exclaim2", new ExclamationBolt(), 2).shuffleGrouping("exclaim1");

    Config conf = new Config();
    conf.setDebug(true);

    if (args != null && args.length > 0) {
        conf.setNumWorkers(30);

        StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
    }
    else {

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("test", conf, builder.createTopology());
        Utils.sleep(5000);
        cluster.killTopology("test");
        cluster.shutdown();
    }
}

}
```

本地运行这个样例，会有类似如下的日志打印，从这个打印中可以看到，Bolt exclaim1的数据来自于Spout word的10个task，即task[7-16]

```
jackson is from task 11 of Spout/Bolt:word
mike is from task 8 of Spout/Bolt:word
nathan is from task 12 of Spout/Bolt:word
nathan is from task 16 of Spout/Bolt:word
nathan is from task 13 of Spout/Bolt:word
```

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.

样例

对上面的样例稍加改造

builder.setSpout("word", new TestWordSpout(), 10);
builder.setBolt("exclaim1", new ExclamationBolt(), 3).fieldsGrouping("word", new Fields("word"));
builder.setBolt("exclaim2", new ExclamationBolt(), 2).shuffleGrouping("exclaim1");

从运行的结果中可以看到类似如下的打印，说明相同的字符都来自于同一个task

mike!!! is from task 2 of Spout/Bolt:exclaim1
mike!!! is from task 2 of Spout/Bolt:exclaim1
mike!!! is from task 2 of Spout/Bolt:exclaim1
mike!!! is from task 2 of Spout/Bolt:exclaim1

或者在execute方法中在加如下的打印System.out.println("Current thread is " + Thread.currentThread().getId() + " to emit " + tuple.getString(0) + "!!!");，可以看到类似如下的打印，所有的mike!!!都是由同一个线程处理的。

Current thread is 124 to emit mike!!!
Current thread is 124 to emit mike!!!

All grouping

定义:The stream is replicated across all the bolt's tasks. Use this grouping with care.

样例

对上面的样例稍加改造

builder.setSpout("word", new TestWordSpout(), 10);
builder.setBolt("exclaim1", new ExclamationBolt(), 3).allGrouping("word");
builder.setBolt("exclaim2", new ExclamationBolt(), 2).shuffleGrouping("exclaim1");

从运行的结果中可以看到类似如下的打印，因为Bolt exclaim1的有3个task，所以下面的结果说明了，Bolt exclaim2要从每个task中都取一次

Current thread is 124 to emit mike!!!
Current thread is 128 to emit mike!!!
Current thread is 150 to emit mike!!!

[Thread-18-exclaim1-executor[2 2]] INFO  o.a.s.d.executor - TRANSFERING tuple [dest: 6 tuple: source: exclaim1:2, stream: default, id: {}, [mike!!!]]
[Thread-22-exclaim1-executor[3 3]] INFO  o.a.s.d.executor - TRANSFERING tuple [dest: 5 tuple: source: exclaim1:3, stream: default, id: {}, [mike!!!]]
[Thread-44-exclaim1-executor[4 4]] INFO  o.a.s.d.executor - TRANSFERING tuple [dest: 6 tuple: source: exclaim1:4, stream: default, id: {}, [mike!!!]]

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.

样例

对上面的样例稍加改造

 builder.setSpout("word", new TestWordSpout(), 10);
 builder.setBolt("exclaim1", new ExclamationBolt(), 3).globalGrouping("word");
 builder.setBolt("exclaim2", new ExclamationBolt(), 2).shuffleGrouping("exclaim1");

结果中会有类似如下的打印，说明mike!!!都来自于了同一个Bolt

mike!!! is from task 2 of Spout/Bolt:exclaim1
mike!!! is from task 2 of Spout/Bolt:exclaim1
mike!!! is from task 2 of Spout/Bolt:exclaim1

None grouping

定义:This grouping specifies that you don't care how the stream is grouped. Currently, none
groupings are equivalent to shuffle groupings.

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.

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.