public static void SetLocalTopology() throws Exception {
TopologyBuilder builder = SetBuilder();
LocalCluster cluster = new LocalCluster();
cluster.submitTopology(topologyName, conf, builder.createTopology());
Thread.sleep(600000);
cluster.shutdown();
}