public static void main(String[] args) throws Exception {

    MemoryTransactionalSpout spout = new MemoryTransactionalSpout(DATA, new Fields("word"), PARTITION_TAKE_PER_BATCH);

    TransactionalTopologyBuilder builder = new TransactionalTopologyBuilder("global-count", "spout", spout, 3);

    builder.setBolt("partial-count", new BatchCount(), 5).noneGrouping("spout");

    builder.setBolt("sum", new UpdateGlobalCount()).globalGrouping("partial-count");



    LocalCluster cluster = new LocalCluster();



    Config config = new Config();

    config.setDebug(true);

    config.setMaxSpoutPending(3);



    cluster.submitTopology("global-count-topology", config, builder.buildTopology());



    Thread.sleep(3000);

    cluster.shutdown();

  }