def jar(jarfile, klass, *args):"""Syntax: [storm jar topology-jar-path class ...]Runs the main method of class with the specified arguments.
The storm jars and configs in ~/.storm are put on the classpath.
The process is configured so that StormSubmitter
(http://nathanmarz.github.com/storm/doc/backtype/storm/StormSubmitter.html)
will upload the jar at topology-jar-path when the topology is submitted.
"""exec_storm_class(klass,jvmtype="-client",extrajars=[jarfile, CONF_DIR, STORM_DIR + "/bin"],args=args,childopts="-Dstorm.jar=" + jarfile)def exec_storm_class(klass, jvmtype="-server", childopts="", extrajars=[], args=[]):nativepath = confvalue("java.library.path", extrajars)args_str = " ".join(map(lambda s: "\"" + s + "\"", args))command = "java" + jvmtype + " -Dstorm.home=" + STORM_DIR + " " + get_config_opts() + " -Djava.library.path=" + nativepath + " " + childopts + " -cp" + get_classpath(extrajars) + " " + klass + " " + args_strprint "Running:" + commandos.system(command)

    public static void main(String[] args) throws Exception {        TopologyBuilder builder = new TopologyBuilder();        builder.setSpout("spout", new RandomSentenceSpout(), 5);        builder.setBolt("split", new SplitSentence(), 8).shuffleGrouping("spout");builder.setBolt("count", new WordCount(), 12).fieldsGrouping("split", new Fields("word"));Config conf = new Config();conf.setDebug(true);if(args!=null && args.length > 0) {conf.setNumWorkers(3);            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());} else {        conf.setMaxTaskParallelism(3);LocalCluster cluster = new LocalCluster();cluster.submitTopology("word-count", conf, builder.createTopology());   Thread.sleep(10000);cluster.shutdown();}}

    /*** Submits a topology to run on the cluster. A topology runs forever or until * explicitly killed.*** @param name the name of the storm.* @param stormConf the topology-specific configuration. See {@link Config}. * @param topology the processing to execute.* @param options to manipulate the starting of the topology* @throws AlreadyAliveException if a topology with this name is already running* @throws InvalidTopologyException if an invalid topology was submitted*/public static void submitTopology(String name, Map stormConf, StormTopology topology, SubmitOptions opts) throws AlreadyAliveException, InvalidTopologyException {if(!Utils.isValidConf(stormConf)) {throw new IllegalArgumentException("Storm conf is not valid. Must be json-serializable");}stormConf = new HashMap(stormConf);stormConf.putAll(Utils.readCommandLineOpts());Map conf = Utils.readStormConfig();conf.putAll(stormConf);try {String serConf = JSONValue.toJSONString(stormConf);if(localNimbus!=null) {LOG.info("Submitting topology " + name + " in local mode");localNimbus.submitTopology(name, null, serConf, topology);} else {NimbusClient client = NimbusClient.getConfiguredClient(conf);if(topologyNameExists(conf, name)) {throw new RuntimeException("Topology with name `" + name + "` already exists on cluster");}submitJar(conf);try {LOG.info("Submitting topology " +  name + " in distributed mode with conf " + serConf);if(opts!=null) {client.getClient().submitTopologyWithOpts(name, submittedJar, serConf, topology, opts);                    } else {// this is for backwards compatibilityclient.getClient().submitTopology(name, submittedJar, serConf, topology);                                            }} catch(InvalidTopologyException e) {LOG.warn("Topology submission exception", e);throw e;} catch(AlreadyAliveException e) {LOG.warn("Topology already alive exception", e);throw e;} finally {client.close();}}LOG.info("Finished submitting topology: " +  name);} catch(TException e) {throw new RuntimeException(e);}}