import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static StormTopology buildTopology(WindowsStoreFactory windowStore, WindowConfig windowConfig) throws Exception {
    TridentKafkaConfig config = new TridentKafkaConfig(new ZkHosts("localhost:2181"), "test");
    config.scheme = new SchemeAsMultiScheme(new StringScheme());
    config.startOffsetTime = kafka.api.OffsetRequest.LatestTime();
    TransactionalTridentKafkaSpout spout = new TransactionalTridentKafkaSpout(config);
    TridentTopology topology = new TridentTopology();

    Stream stream = topology.newStream("spout1", spout).each(new Fields("str"),
            new Split(), new Fields("word"))
            .window(windowConfig, windowStore, new Fields("word"), new CountAsAggregator(), new Fields("count"))
            .peek(new Consumer() {
                @Override
                public void accept(TridentTuple tuple) {
		System.out.print("[");
		for (int index = 0; index < tuple.size(); index++) {
			System.out.print(tuple.get(index));
			if (index < (tuple.size() - 1))
				System.out.print(",");
		}
		System.out.println("]");
	}
            });

    return topology.build();
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static void main(String args[]) {
	TridentKafkaConfig config = new TridentKafkaConfig(new ZkHosts(
			"localhost:2181"), "storm-trident-diy");
	config.scheme = new SchemeAsMultiScheme(new StringScheme());
	config.startOffsetTime = kafka.api.OffsetRequest.LatestTime();
	TransactionalTridentKafkaSpout spout = new TransactionalTridentKafkaSpout(
			config);
	TridentTopology topology = new TridentTopology();
	topology.newStream("spout", spout).filter(new ExcludePhoneNumber())
			.each(new Fields("str"), new DeviceInfoExtractor(), new Fields("phone", "bytes"))
			.groupBy(new Fields("phone"))
			.aggregate(new Fields("bytes", "phone"), new Sum(), new Fields("sum"))
			.applyAssembly(new FirstN(10, "sum"))
			.each(new Fields("phone", "sum"), new Debug());
	
	Config config1 = new Config();
	config1.setNumWorkers(3);
	LocalCluster cluster = new LocalCluster();
	cluster.submitTopology("storm-trident-diy", config1, topology.build());
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static void main(String args[]) throws InterruptedException {
	Config config = new Config();
	config.setNumWorkers(3);
	TopologyBuilder topologyBuilder = new TopologyBuilder();
	
	String zkConnString = "localhost:2181";
	String topicName = "sensor-data";
	
	BrokerHosts hosts = new ZkHosts(zkConnString);
	SpoutConfig spoutConfig = new SpoutConfig(hosts, topicName , "/" + topicName, UUID.randomUUID().toString());
	spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

	KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
	topologyBuilder.setSpout("spout", kafkaSpout, 1);
       topologyBuilder.setBolt("es-bolt", new ElasticSearchBolt(), 1).shuffleGrouping("spout");
       
       LocalCluster cluster = new LocalCluster();
       cluster.submitTopology("storm-es-example", config, topologyBuilder.createTopology());
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static void main(String[] args) {
	Config config = new Config();
	config.setNumWorkers(3);
	TopologyBuilder topologyBuilder = new TopologyBuilder();
	
	String zkConnString = "localhost:2181";
	String topicName = "storm-diy";
	
	BrokerHosts hosts = new ZkHosts(zkConnString);
	SpoutConfig spoutConfig = new SpoutConfig(hosts, topicName , "/" + topicName, UUID.randomUUID().toString());
	spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

	KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
	topologyBuilder.setSpout("spout", kafkaSpout, 1);
	topologyBuilder.setBolt("parser", new ParseAndUsageBolt(), 1).shuffleGrouping("spout");
	topologyBuilder.setBolt("usageCassandra", new UsageCassandraBolt("localhost", "usage"), 1).shuffleGrouping("parser", "usagestream");
	topologyBuilder.setBolt("tdrCassandra", new TDRCassandraBolt("localhost", "tdr"), 1).shuffleGrouping("parser", "tdrstream");
	
	LocalCluster cluster = new LocalCluster();
       cluster.submitTopology("storm-diy", config, topologyBuilder.createTopology());
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
private SpoutConfig constructKafkaSpoutConf() {
    // BrokerHosts hosts = new ZkHosts(topologyConfig.getProperty("kafka.zookeeper.host.port"));
    BrokerHosts hosts = new ZkHosts("localhost:2181");
    /*
    String topic = topologyConfig.getProperty("kafka.topic");
    String zkRoot = topologyConfig.getProperty("kafka.zkRoot");
    String consumerGroupId = topologyConfig.getProperty("kafka.consumer.group.id");
    */
    String topic = "addresses";
    String zkRoot = "";
    String consumerGroupId = "group1";

    SpoutConfig spoutConfig = new SpoutConfig(hosts, topic, zkRoot, consumerGroupId);
    spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

    return spoutConfig;
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
/**
 * Creates Kafka spout.
 *
 * @param topic Kafka topic
 * @return {@link KafkaSpout}
 */
protected org.apache.storm.kafka.KafkaSpout createKafkaSpout(String topic, String spoutId) {
    String zkRoot = String.format("/%s/%s", getTopologyName(), topic);
    ZkHosts hosts = new ZkHosts(config.getZookeeperHosts());

    SpoutConfig cfg = new SpoutConfig(hosts, topic, zkRoot, spoutId);
    cfg.startOffsetTime = OffsetRequest.EarliestTime();
    cfg.scheme = new SchemeAsMultiScheme(new StringScheme());
    cfg.bufferSizeBytes = 1024 * 1024 * 4;
    cfg.fetchSizeBytes = 1024 * 1024 * 4;

    return new org.apache.storm.kafka.KafkaSpout(cfg);
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
/**
 * Extract spout config from Topology configuration.
 * @param props    The topology configuration.
 * @return         An SpoutConfig object
 */
private static SpoutConfig extractSpoutConfig(Properties props, String topic)
    throws ClassNotFoundException, IllegalAccessException, InstantiationException {

  // Zookeeper hosts
  ZkHosts zkHosts = new ZkHosts(props.getProperty("zookeeper.connect"));

  // Client ID
  String clientId = props.getProperty("consumer.id");
  clientId = clientId.length() > 0 ?  clientId : UUID.randomUUID().toString();

  // Spout config
  SpoutConfig conf = new SpoutConfig(zkHosts, topic, "/kafka-mirror/" + topic, clientId);

  // Get scheme classname from properties or use StringScheme by default
  String schemeClassname = (String) props.getOrDefault("spout.scheme",
                                                       "org.apache.storm.kafka.StringScheme");

  Class<?> clazz = Class.forName(schemeClassname);
  conf.scheme = new SchemeAsMultiScheme((Scheme) clazz.newInstance());
  return conf;
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static void runPirkTopology() throws PIRException
{
  // Set up Kafka parameters
  logger.info("Configuring Kafka.");
  String zkRoot = "/" + kafkaTopic + "_pirk_storm";
  BrokerHosts zkHosts = new ZkHosts(brokerZk);
  SpoutConfig kafkaConfig = new SpoutConfig(zkHosts, kafkaTopic, zkRoot, kafkaClientId);
  kafkaConfig.ignoreZkOffsets = forceFromStart;

  // Create conf
  logger.info("Retrieving Query and generating Storm conf.");
  Config conf = createStormConf();
  Query query = StormUtils.getQuery(useHdfs, hdfsUri, queryFile);
  conf.put(StormConstants.N_SQUARED_KEY, query.getNSquared().toString());
  conf.put(StormConstants.QUERY_INFO_KEY, query.getQueryInfo().toMap());

  // Configure this for different types of input data on Kafka.
  kafkaConfig.scheme = new SchemeAsMultiScheme(new PirkHashScheme(conf));

  // Create topology
  StormTopology topology = getPirkTopology(kafkaConfig);

  // Run topology
  logger.info("Submitting Pirk topology to Storm...");
  try
  {
    StormSubmitter.submitTopologyWithProgressBar(topologyName, conf, topology);
  } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e)
  {
    throw new PIRException(e);
  }

}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
private SpoutConfig setUpTestKafkaSpout(Config conf)
{
  ZkHosts zkHost = new ZkHosts(zookeeperLocalCluster.getConnectString());

  SpoutConfig kafkaConfig = new SpoutConfig(zkHost, topic, "/pirk_test_root", "pirk_integr_test_spout");
  kafkaConfig.scheme = new SchemeAsMultiScheme(new PirkHashScheme(conf));
  logger.info("KafkaConfig initialized...");

  return kafkaConfig;
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static void main(String[] args)
		throws InterruptedException, InvalidTopologyException, AuthorizationException, AlreadyAliveException {

	String topologyName = "TSAS";// 元组名
	// Zookeeper主机地址，会自动选取其中一个
	ZkHosts zkHosts = new ZkHosts("192.168.230.128:2181,192.168.230.129:2181,192.168.230.131:2181");
	String topic = "trademx";
	String zkRoot = "/storm";// storm在Zookeeper上的根路径
	String id = "tsaPro";

	// 创建SpoutConfig对象
	SpoutConfig spontConfig = new SpoutConfig(zkHosts, topic, zkRoot, id);

	TopologyBuilder builder = new TopologyBuilder();
	builder.setSpout("kafka", new KafkaSpout(spontConfig), 2);
	builder.setBolt("AccBolt", new AccBolt()).shuffleGrouping("kafka");
	builder.setBolt("ToDbBolt", new ToDbBolt()).shuffleGrouping("AccBolt");

	Config config = new Config();
	config.setDebug(false);

	if (args.length == 0) { // 本地运行，用于测试
		LocalCluster localCluster = new LocalCluster();
		localCluster.submitTopology(topologyName, config, builder.createTopology());
		Thread.sleep(1000 * 3600);
		localCluster.killTopology(topologyName);
		localCluster.shutdown();
	} else { // 提交至集群运行
		StormSubmitter.submitTopology(topologyName, config, builder.createTopology());
	}

}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
private SpoutConfig constructKafkaSpoutConf() {
    // BrokerHosts hosts = new ZkHosts(topologyConfig.getProperty("kafka.zookeeper.host.port"));
    BrokerHosts hosts = new ZkHosts("localhost:2181");

    String topic = "properties";
    String zkRoot = "";
    String consumerGroupId = "group1";

    SpoutConfig spoutConfig = new SpoutConfig(hosts, topic, zkRoot, consumerGroupId);
    spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

    return spoutConfig;
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static SpoutConfig getSpoutConfig(Map options, MultiScheme scheme) throws IllegalArgumentException {
  String zkServers = (String) Utils.get(options, ZOOKEEPER_SERVERS, "localhost:2181");
  String kafkaRoot = (String) Utils.get(options, KAFKA_ROOT_PATH, "/kafka");
  String connectString = zkServers + kafkaRoot;

  BrokerHosts hosts = new ZkHosts(connectString);
  String topic = (String) Utils.get(options, TOPIC, DEFAULT_TOPIC);
  String appId = (String) Utils.get(options, CLIENT_ID, "storm-app");

  SpoutConfig config = new SpoutConfig(hosts, topic, kafkaRoot, appId);
  //config.forceFromStart = true;
  config.zkServers = new ArrayList<String>();

  String [] servers = zkServers.split(",");

  for (int i = 0; i < servers.length; i++) {
    String[] serverAndPort = servers[0].split(":");
    config.zkServers.add(serverAndPort[0]);
    int port = Integer.parseInt(serverAndPort[1]);
    if (i == 0) {
      config.zkPort = port;
    }

    if (config.zkPort != port) {
      throw new IllegalArgumentException("The zookeeper port on all  server must be same");
    }
  }
  config.scheme = scheme;
  return config;
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static TridentKafkaConfig getTridentKafkaConfig(Map options, MultiScheme scheme) {
  String zkServers = (String) Utils.get(options, ZOOKEEPER_SERVERS, "localhost:2181") ;
  String kafkaRoot = (String) Utils.get(options, KAFKA_ROOT_PATH, "/kafka");
  String connectString = zkServers + kafkaRoot;

  BrokerHosts hosts = new ZkHosts(connectString);
  String topic = (String) Utils.get(options, TOPIC, DEFAULT_TOPIC);
  String appId = (String) Utils.get(options, CLIENT_ID, "storm-app");

  TridentKafkaConfig config = new TridentKafkaConfig(hosts, topic, appId);
  config.scheme = scheme;
  return config;
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
/**
 * Creates a transactional kafka spout that consumes any new data published to "test" topic.
 * <p/>
 * For more info on transactional spouts
 * see "Transactional spouts" section in
 * <a href="https://storm.apache.org/documentation/Trident-state"> Trident state</a> doc.
 *
 * @return a transactional trident kafka spout.
 */
private TransactionalTridentKafkaSpout createKafkaSpout() {
    ZkHosts hosts = new ZkHosts(zkUrl);
    TridentKafkaConfig config = new TridentKafkaConfig(hosts, "test");
    config.scheme = new SchemeAsMultiScheme(new StringScheme());

    // Consume new data from the topic
    config.startOffsetTime = kafka.api.OffsetRequest.LatestTime();
    return new TransactionalTridentKafkaSpout(config);
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static void main(String[] args) {
	
	if(args.length <1){
		System.out.println("Please mention deployment mode either local or cluster");
		System.exit(1);
	}
	
	String deploymentMode = args[0];
	
	Config config = new Config();
	config.setNumWorkers(3);
	TopologyBuilder topologyBuilder = new TopologyBuilder();
	
	String zkConnString = "localhost:2181";
	String topicName = "vehicle-data";
	String hcHostName = "localhost";
	String hcPort = "5701";
	String esClusterName = "cluster.name";
	String esApplicationName = "my-application";
	String esHostName = "localhost";
	int esPort = 9300;
	
	BrokerHosts hosts = new ZkHosts(zkConnString);
	SpoutConfig spoutConfig = new SpoutConfig(hosts, topicName , "/" + topicName, UUID.randomUUID().toString());
	spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

	KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
	topologyBuilder.setSpout("spout", kafkaSpout, 1);
	topologyBuilder.setBolt("parser", new ParseBolt(), 1).shuffleGrouping("spout");
	topologyBuilder.setBolt("checkAndAlert", new CheckDistanceAndAlertBolt(hcHostName, hcPort), 1).shuffleGrouping("parser","parsedstream");
	topologyBuilder.setBolt("saveTDR", new ElasticSearchBolt("vehicle-tdr", "tdr",esClusterName, esApplicationName,esHostName, esPort),1).shuffleGrouping("parser","parsedstream");
	topologyBuilder.setBolt("generateAlert", new GenerateAlertBolt(hcHostName, hcPort), 1).shuffleGrouping("checkAndAlert", "alertInfo");
	topologyBuilder.setBolt("saveAlert", new ElasticSearchBolt("vehicle-alert", "alert",esClusterName, esApplicationName,esHostName, esPort), 1).shuffleGrouping("generateAlert", "generatedAlertInfo");
	
	LocalCluster cluster = new LocalCluster();
	if (deploymentMode.equalsIgnoreCase("local")) {
		System.out.println("Submitting topology on local");
		cluster.submitTopology(topicName, config, topologyBuilder.createTopology());
	} else {
        try {
        	System.out.println("Submitting topology on cluster");
			StormSubmitter.submitTopology(topicName, config, topologyBuilder.createTopology());
		} catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
			e.printStackTrace();
		}
	}
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {
    String zks = "root2:2181,root4:2181,root5:2181";// default zookeeper configuration
    String topic= "tweetswordtopic3";// default kafka topic configuration
    String zkRoot = "/stormkafka"; // default zookeeper root configuration for storm
    String id = "DStreamTopology";// default application ID

    BrokerHosts brokerHosts = new ZkHosts(zks,"/kafka/brokers");// default kafka BrokerHosts
    SpoutConfig spoutConf = new SpoutConfig(brokerHosts, topic, zkRoot, id);
    spoutConf.scheme = new SchemeAsMultiScheme(new MyScheme());
    spoutConf.ignoreZkOffsets = true;
    spoutConf.zkServers = Arrays.asList(new String[] {"root2", "root4", "root5"});
    spoutConf.zkPort = 2181;
    //      spoutConf.startOffsetTime = kafka.api.OffsetRequest.LatestTime();
    spoutConf.startOffsetTime = kafka.api.OffsetRequest.EarliestTime();
    KafkaSpout kafkaSpout=new KafkaSpout(spoutConf);

    WordCounterBolt wordCounterBolt=new WordCounterBolt();
    MyAggregatorBolt aggregatorBolt=new MyAggregatorBolt();

    SchedulingTopologyBuilder builder=new SchedulingTopologyBuilder();
    Integer numworkers=Integer.valueOf(args[1]);

    builder.setSpout(KAFKA_SPOUT_ID, kafkaSpout, 9);
    builder.setBalancingScheduling(KAFKA_SPOUT_ID,"word");
    builder.setBolt(WORDCOUNTER_BOLT_ID,wordCounterBolt, 36).fieldsGrouping(SPLITTER_BOLT_ID+builder.getSchedulingNum(), Constraints.nohotFileds, new Fields(Constraints.wordFileds)).shuffleGrouping(SPLITTER_BOLT_ID+builder.getSchedulingNum(), Constraints.hotFileds);
    builder.setBolt(AGGREGATOR_BOLT_ID, aggregatorBolt, 36).fieldsGrouping(WORDCOUNTER_BOLT_ID, new Fields(Constraints.wordFileds));
    //Topology config
    Config config=new Config();
    config.setNumWorkers(numworkers);//config numworkers
    if(args[0].equals("local")){
        LocalCluster localCluster=new LocalCluster();

        localCluster.submitTopology(TOPOLOGY_NAME,config,builder.createTopology());
        Utils.sleep(50*1000);//50s
        localCluster.killTopology(TOPOLOGY_NAME);
        localCluster.shutdown();
    }else {
        StormSubmitter.submitTopology(args[0],config,builder.createTopology());
    }

}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static void main(String[] args) {
	// zookeeper hosts for the Kafka cluster
	BrokerHosts zkHosts = new ZkHosts("localhost:2181");

	// Create the KafkaReadSpout configuartion
	// Second argument is the topic name
	// Third argument is the zookeeper root for Kafka
	// Fourth argument is consumer group id
	SpoutConfig kafkaConfig = new SpoutConfig(zkHosts, "dataTopic", "",
			"id7");

	// Specify that the kafka messages are String
	kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

	// We want to consume all the first messages in the topic everytime
	// we run the topology to help in debugging. In production, this
	// property should be false
	kafkaConfig.startOffsetTime = kafka.api.OffsetRequest.EarliestTime();

	// Now we create the topology
	TopologyBuilder builder = new TopologyBuilder();

	// set the kafka spout class
	builder.setSpout("KafkaReadSpout", new KafkaSpout(kafkaConfig), 1);

	// use "|" instead of "," for field delimiter
	RecordFormat format = new DelimitedRecordFormat()
			.withFieldDelimiter(",");

	// sync the filesystem after every 1k tuples
	SyncPolicy syncPolicy = new CountSyncPolicy(1000);

	// rotate files when they reach 5MB
	FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f,
			Units.MB);

	FileNameFormat fileNameFormatHDFS = new DefaultFileNameFormat()
			.withPath("/hdfs-bolt-output/");

	HdfsBolt hdfsBolt2 = new HdfsBolt().withFsUrl("hdfs://127.0.0.1:8020")
			.withFileNameFormat(fileNameFormatHDFS)
			.withRecordFormat(format).withRotationPolicy(rotationPolicy)
			.withSyncPolicy(syncPolicy);

	//
	builder.setBolt("HDFS2", hdfsBolt2).shuffleGrouping("KafkaReadSpout");

	// create an instance of LocalCluster class for executing topology in
	// local mode.
	LocalCluster cluster = new LocalCluster();
	Config conf = new Config();

	// Submit topology for execution
	cluster.submitTopology("KafkaHadoop", conf, builder.createTopology());

	try {
		// Wait for some time before exiting
		System.out.println("Waiting to consume from kafka");
		Thread.sleep(6000000);
	} catch (Exception exception) {
		System.out.println("Thread interrupted exception : " + exception);
	}

	// kill the KafkaTopology
	cluster.killTopology("KafkaToplogy");

	// shut down the storm test cluster
	cluster.shutdown();

}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static void main(String[] args) {
	// zookeeper hosts for the Kafka cluster
	BrokerHosts zkHosts = new ZkHosts("localhost:2181");

	// Create the KafkaSpout configuartion
	// Second argument is the topic name
	// Third argument is the zookeeper root for Kafka
	// Fourth argument is consumer group id
	SpoutConfig kafkaConfig = new SpoutConfig(zkHosts, "twitterData", "",
			"id7");

	// Specify that the kafka messages are String
	kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

	// We want to consume all the first messages in the topic everytime
	// we run the topology to help in debugging. In production, this
	// property should be false
	kafkaConfig.startOffsetTime = kafka.api.OffsetRequest
			.EarliestTime();

	// Now we create the topology
	TopologyBuilder builder = new TopologyBuilder();

	// set the kafka spout class
	builder.setSpout("KafkaSpout", new KafkaSpout(kafkaConfig), 1);

	// use "|" instead of "," for field delimiter
	RecordFormat format = new DelimitedRecordFormat()
			.withFieldDelimiter(",");

	// sync the filesystem after every 1k tuples
	SyncPolicy syncPolicy = new CountSyncPolicy(1000);

	// rotate files when they reach 5MB
	FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f,
			Units.MB);

	FileNameFormat fileNameFormatSentiment = new DefaultFileNameFormat()
	.withPath("/sentiment-tweet/");

	HdfsBolt hdfsBolt2 = new HdfsBolt().withFsUrl("hdfs://127.0.0.1:8020")
			.withFileNameFormat(fileNameFormatSentiment).withRecordFormat(format)
			.withRotationPolicy(rotationPolicy).withSyncPolicy(syncPolicy);


	//builder.setBolt("HDFSBolt", hdfsBolt).shuffleGrouping("KafkaSpout");
	builder.setBolt("json", new JSONParsingBolt()).shuffleGrouping("KafkaSpout");
	
	//
	builder.setBolt("sentiment", new SentimentBolt("/home/centos/Desktop/workspace/storm_twitter/src/main/resources/AFINN-111.txt")).shuffleGrouping("json","stream2");

	//
	builder.setBolt("HDFS2", hdfsBolt2).shuffleGrouping("sentiment");

	// create an instance of LocalCluster class for executing topology in
	// local mode.
	LocalCluster cluster = new LocalCluster();
	Config conf = new Config();

	// Submit topology for execution
	cluster.submitTopology("KafkaToplogy", conf, builder.createTopology());

	try {
		// Wait for some time before exiting
		System.out.println("Waiting to consume from kafka");
		Thread.sleep(6000000);
	} catch (Exception exception) {
		System.out.println("Thread interrupted exception : " + exception);
	}

	// kill the KafkaTopology
	cluster.killTopology("KafkaToplogy");

	// shut down the storm test cluster
	cluster.shutdown();

}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static void main(String[] args) {
	try {
		// zookeeper hosts for the Kafka cluster
		BrokerHosts zkHosts = new ZkHosts("10.191.208.89:2183");

		// Create the KafkaSpout configuartion
		// Second argument is the topic name
		// Third argument is the zookeepr root for Kafka
		// Fourth argument is consumer group id
		SpoutConfig kafkaConfig = new SpoutConfig(zkHosts, "new_topic", "",
				"id1");

		// Specify that the kafka messages are String
		// We want to consume all the first messages in the topic everytime
		// we run the topology to help in debugging. In production, this
		// property should be false
		kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
		kafkaConfig.startOffsetTime = kafka.api.OffsetRequest
				.EarliestTime();

		// Now we create the topology
		TopologyBuilder builder = new TopologyBuilder();

		// set the kafka spout class
		builder.setSpout("KafkaSpout", new KafkaSpout(kafkaConfig), 2);

		// set the word and sentence bolt class
		builder.setBolt("WordBolt", new WordBolt(), 1).globalGrouping(
				"KafkaSpout");
		builder.setBolt("SentenceBolt", new SentenceBolt(), 1)
				.globalGrouping("WordBolt");

		// create an instance of LocalCluster class for executing topology
		// in local mode.
		LocalCluster cluster = new LocalCluster();
		Config conf = new Config();
		conf.setDebug(true);
		if (args.length > 0) {
			conf.setNumWorkers(2);
			conf.setMaxSpoutPending(5000);
			StormSubmitter.submitTopology("KafkaToplogy1", conf,
					builder.createTopology());

		} else {
			// Submit topology for execution
			cluster.submitTopology("KafkaToplogy1", conf,
					builder.createTopology());
			System.out.println("called1");
			Thread.sleep(1000000);
			// Wait for sometime before exiting
			System.out.println("Waiting to consume from kafka");

			System.out.println("called2");
			// kill the KafkaTopology
			cluster.killTopology("KafkaToplogy1");
			System.out.println("called3");
			// shutdown the storm test cluster
			cluster.shutdown();
		}

	} catch (Exception exception) {
		System.out.println("Thread interrupted exception : " + exception);
	}
}
 

import org.apache.storm.kafka.ZkHosts; //导入依赖的package包/类
public static void main(String[] args) throws Exception{
    Config config = new Config();
    IRedisDB _redis = RedisDB.getInstance("127.0.0.1", 6379);
    config.setDebug(true);
    config.put("mongodb.ip", "127.0.0.1");
    config.put("mongodb.port", 27017);
    config.put("redis.ip", "127.0.0.1");
    config.put("redis.port", 6379);
    config.put("kafka.properties", "127.0.0.1:9092");
    config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
    String zkConnString = "localhost:2181";
    BrokerHosts brokerHosts = new ZkHosts(zkConnString);

    SpoutConfig eventConfig = new SpoutConfig(brokerHosts, "event", "/event","storm");
    eventConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
    eventConfig.startOffsetTime = -1;
    
    SpoutConfig proceedConfig = new SpoutConfig(brokerHosts, "proceed", "/proceed", "storm");
    proceedConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
    proceedConfig.startOffsetTime = -1;
    
    SpoutConfig orderConfig = new SpoutConfig(brokerHosts, "order", "/order", "storm");
    orderConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
    orderConfig.startOffsetTime = -1;
    
    String topicTrigger = "trigger";
    SpoutConfig triggerConfig = new SpoutConfig(brokerHosts, topicTrigger, "/"+topicTrigger, "storm");
    triggerConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
    triggerConfig.startOffsetTime = -1;
    
    String topicStatus = "status";
    SpoutConfig statusConfig = new SpoutConfig(brokerHosts, topicStatus, "/"+topicStatus, "storm");
    statusConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
    statusConfig.startOffsetTime = -1;
    
    TopologyBuilder builder = new TopologyBuilder();
    
    builder.setSpout("event-spout", new KafkaSpout(eventConfig));
    builder.setSpout("proceed-spout", new KafkaSpout(proceedConfig));
    builder.setSpout("order-spout", new KafkaSpout(orderConfig));
    builder.setBolt("indexing-bolt", new IndexingBolt()).allGrouping("event-spout").allGrouping("proceed-spout").allGrouping("order-spout");
    builder.setBolt("staging-bolt", new StagingBolt()).fieldsGrouping("indexing-bolt", new Fields("roadMapId"));
    builder.setBolt("calling-trigger-bolt", new CallingTriggerBolt()).fieldsGrouping("staging-bolt", new Fields("roadMapId"));

    builder.setSpout("trigger-spout", new KafkaSpout(triggerConfig));
    builder.setSpout("status-spout", new KafkaSpout(statusConfig));
    builder.setBolt("scheduling-bolt", new SchedulingBolt())
            .shuffleGrouping("trigger-spout");
    builder.setBolt("status-bolt", new StatusBolt())
            .shuffleGrouping("status-spout");
    builder.setBolt("executing-bolt", new ExecutingBolt()).fieldsGrouping("scheduling-bolt",new Fields("roadMapId"));
    builder.setBolt("provisioning-bolt", new ProvisioningBolt()).fieldsGrouping("executing-bolt",new Fields("roadMapId"));
    builder.setBolt("calling-feed-bolt", new CallingFeedBolt()).fieldsGrouping("provisioning-bolt",new Fields("roadMapId"));
    _redis.deleteAllNodes();
    LocalCluster cluster = new LocalCluster();
    cluster.submitTopology("TriggerTopology", config, builder.createTopology());

}