Storm的分组方式

Storm中内置了7种分组方式

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.