public static void main(String[] args) throws Exception {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new RandomIntegerSpout());
builder.setBolt("sumbolt", new WindowSumBolt().withWindow(new Count(5), new Count(3))
.withMessageIdField("msgid"), 1).shuffleGrouping("spout");
builder.setBolt("printer", new PrinterBolt(), 1).shuffleGrouping("sumbolt");
Config conf = new Config();
conf.setDebug(false);
if (args != null && args.length > 0) {
conf.setNumWorkers(1);
StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
} else {
LocalCluster cluster = new LocalCluster();
StormTopology topology = builder.createTopology();
cluster.submitTopology("test", conf, topology);
Utils.sleep(40000);
cluster.killTopology("test");
cluster.shutdown();
}
}
|