global分组
对目标target tasked进行排序,选择最小的taskId号进行发送tuple
类似于direct,可以是特殊的direct分组。
package com.mao.storm.group.global;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
/**
* global是direct分组的特例,选择最小的taskId
*/
public class App {
public static void main(String[] args) throws Exception {
TopologyBuilder builder = new TopologyBuilder();
//设置Spout
builder.setSpout("wcspout", new WordCountSpout()).setNumTasks(2);
//设置creator-Bolt
builder.setBolt("split-bolt", new SplitBolt(),2).noneGrouping("wcspout").setNumTasks(2);
Config conf = new Config();
conf.setNumWorkers(2);
conf.setDebug(true);
/**
* 本地模式storm
*/
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("wc", conf, builder.createTopology());
System.out.println("hello world");
//Thread.sleep(20000);
// StormSubmitter.submitTopology("wordcount", conf, builder.createTopology());
//cluster.shutdown();
}
}
package com.mao.storm.group.global;
import com.mao.storm.util.Util;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import java.util.Map;
/**
*/
public class SplitBolt implements IRichBolt {
private TopologyContext context ;
private OutputCollector collector ;
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.context = context ;
this.collector = collector ;
}
public void execute(Tuple tuple) {
String line = tuple.getString(0);
Util.sendToLocalhost(this,line);
String[] arr = line.split(" ");
for(String s : arr){
collector.emit(new Values(s,1));
}
}
public void cleanup() {
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word","count"));
}
public Map<String, Object> getComponentConfiguration() {
return null;
}
}
package com.mao.storm.group.global;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
/**
* Spout
*/
public class WordCountSpout implements IRichSpout{
private TopologyContext context ;
private SpoutOutputCollector collector ;
private List<String> states ;
private Random r = new Random();
private int index = 0;
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.context = context ;
this.collector = collector ;
states = new ArrayList<String>();
states.add("hello world tom");
states.add("hello world tomas");
states.add("hello world tomasLee");
states.add("hello world tomson");
}
public void close() {
}
public void activate() {
}
public void deactivate() {
}
public void nextTuple() {
if(index < 3){
String line = states.get(r.nextInt(4));
collector.emit(new Values(line),index);
index ++ ;
}
}
public void ack(Object msgId) {
}
public void fail(Object msgId) {
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("line"));
}
public Map<String, Object> getComponentConfiguration() {
return null;
}
}