在详细讲解Storm不同组件之间的tuples传递之前，我们先看下我们的结构图

从上面的结构图，我们可以看到。一个topology是spouts和bolts组成的图，而Spout与Blot以及Blot与Blot之间的传递是通过Stream Grouping来完成的。

定义一个topology的其中一步是定义每个bolt接收什么样的流作为输入。stream grouping就是用来定义一个stream应该如果分配数据给bolts上面的多个tasks。

当我们要定义Topology的时候，需要设置Stream Grouping来指定处理流程；

...
    builder.setBlot("word-normalizer",new WordNormalizer()).
<span style="white-space:pre">	</span>shuffleGrouping("word-reader")
...

1、Shuffle Grouping: 随机分组， 也是我们最常用的分组；随机派发stream里面的tuple，保证每个bolt接收到的tuple数目大致相同。我们图片中使用的就是ShuffleGrouping。配置方法很简单，只需指定你上一级流入组件的名称即可；

2、Fields Grouping：按字段分组， 指定某个字段作为分组条件，那么字段值相同的元组（tuple）会被分到同一个Blots里的task进行处理，反之则分配到不同的Blot的task进行处理

在我之前写的wordcount例子中，当你指定流组字段是word，那么word-normalizer将会发送一个指定的word到相同的word-counter Blot实例；

...
    builder.setBlot("word-normalizer",new WordNormalizer(),2).
<span style="white-space:pre">	</span>fieldsGrouping("word-reader",new Fields("word"))
...

注：在Fields Grouping中的所有字段，必须在declareOutputFields方法中存在。否则会报错；

3、All Grouping：顾名思义，就是对于每一个tuple，所有的bolts都会收到。类似于我们的广播；

4、Direct Grouping：直接分组；意思是说，消息生产者直接讲元组指定分配给消息消费者中的某个task进行处理；只有被声明为Direct Stream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发送，而不是emit。消息处理者可以通过TopologyContext来获取处理它的消息的task的id（OutputCollector.emit方法也会返回task的id）；看下下面的例子

public void execute(Tuple input)
    ...
    for(String word:words){
          ...
          collector.emitDirect(getWordCountIndex(word),new Values(word))
    }
     collector.ack(input);
}

这里使用的是emitDirect方法，而不是emit方法

而且我们在Blot的prepare方法中需直声明指定有哪个组件task来接收；这里指定了word-counter

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

在Topology定义的时候是这样定义的；

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

5、Global Grouping：全局分组；全局分组发送的所有实例的源产生一个单一的目标元组实例（具体的说，是发送到ID最低的task进行处理）。

6、None Grouping：不分组；这种分组方法和第一种shuffle Grouping效果是一样的；换句话说，这种方法不关心如何分组；

7、Custom Grouping：自定义分组

除了以上几种分组方式之外，我们还可以自己自定义分组，自己DIY去控制哪些Blot来接收Tuple；

实现自定义分组需要继承backtype.storm.grouping.CustomStreamGrouping 接口；

public class ModuleGrouping implements CustomStreamGrouping, Serializable{
  int numTasks = 0;
  @Override
  public List<Integer> chooseTasks(List<Object> values) {
    List<Integer> boltIds = new ArrayList();
   <span style="white-space:pre">	</span>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();
  }
}

这里写了一个简单的自定义分组（CustomGrouping）,通过对第一个character进行取模来选择由哪个Blot接收数据进行处理；

使用自定义分组的实例如下：

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

参考文献：https://github.com/nathanmarz/storm/wiki/Tutorial

http://blog.linezing.com/category/storm-quick-startspm=0.0.0.0.vUIENl

转载请注明来源地址：http://blog.csdn.net/weijonathan/article/details/18408535