Группировка в простой топологии агрегационной шторма

Я пытаюсь написать топологию, которая делает следующее:

• Носик, который подписывается на канал Twitter (на основе ключевого слова)
• Агрегирующий болт, который объединяет несколько твитов (скажем, N) в коллекции и отправляет им болт принтера
• Простой болт, который сразу выводит коллекцию на консоль.

На самом деле я хочу сделать еще немного обработки в коллекции.

Я тестировал его локально и выглядел так, как будто он работает. Тем не менее, я не уверен, правильно ли я установил группировки на болтах, и если это будет корректно работать при развертывании в реальном кластере штормов. Я был бы признателен, если кто-то сможет помочь в рассмотрении этой топологии и предложить любые ошибки, изменения или улучшения.

Спасибо.

Вот как выглядит моя топология.

builder.setSpout("spout", new TwitterFilterSpout("pittsburgh"));
   builder.setBolt("sampleaggregate", new SampleAggregatorBolt())
                .shuffleGrouping("spout");
   builder.setBolt("printaggreator",new PrinterBolt()).shuffleGrouping("sampleaggregate");

Болт агрегации

public class SampleAggregatorBolt implements IRichBolt {

    protected OutputCollector collector;
    protected Tuple currentTuple;
    protected Logger log;
    /**
     * Holds the messages in the bolt till you are ready to send them out
     */
    protected List<Status> statusCache;

    @Override
    public void prepare(Map stormConf, TopologyContext context,
                        OutputCollector collector) {
        this.collector = collector;

        log = Logger.getLogger(getClass().getName());
        statusCache = new ArrayList<Status>();
    }

    @Override
    public void execute(Tuple tuple) {
        currentTuple = tuple;

        Status currentStatus = null;
        try {
            currentStatus = (Status) tuple.getValue(0);
        } catch (ClassCastException e) {
        }
        if (currentStatus != null) {

            //add it to the status cache
            statusCache.add(currentStatus);
            collector.ack(tuple);


            //check the size of the status cache and pass it to the next stage if you have enough messages to emit
            if (statusCache.size() > 10) {
                collector.emit(new Values(statusCache));
            }

        }
    }

    @Override
    public void cleanup() {


    }

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

    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;  //To change body of implemented methods use File | Settings | File Templates.
    }


    protected void setupNonSerializableAttributes() {

    }

}

Болт принтера

public class PrinterBolt extends BaseBasicBolt {

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        System.out.println(tuple.size() + " "  + tuple);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer ofd) {
    }

}

Ответы

builder.setBolt("sampleaggregate", new SampleAggregatorBolt())
            .shuffleGrouping("spout", new Fields("keyword"));