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 such as a math operation. However, if the operation can’t be randomically distributed, such as the example in Chapter 2 where you needed to count words, you should consider the use of other grouping.

Fields Grouping allows you 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 you 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"));
...

All Grouping sends a single copy of each tuple to all instances of the receiving bolt. This kind of grouping is used to send signals to bolts. For example, if you need to refresh a cache, you can send a refresh cache signal to all bolts. In the word-count example, you 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 gives us the possibility to declare named streams (if you don’t send a tuple to a named stream, the stream is "default"); it’s an excellent way to identify the source of the tuples, such as this case where we want to identify the signals.

In the topology definition, you’ll 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 the git repository.

You can create your own custom stream grouping by implementing the backtype.storm.grouping.CustomStreamGrouping interface. This gives you 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();
    }
}

You 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 the example, change the word-normalizer grouping by the following:

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

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

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, specify that the stream will be grouped directly:

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

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

At the time of this writing (Storm version 0.7.1), using this grouping is the same as using Shuffle Grouping. In other words, when using this grouping, it doesn’t matter how streams are grouped.