storm_WordCount案例

模拟一片文章，计算其个单词出现的次数

【Spout】数据源

public class WordSpout implements IRichSpout {
    private static final long serialVersionUID = 1l;
    private SpoutOutputCollector collector;
    private int index = 0 ;
    //模拟静态数据
    private String[] sentences = {
            "my dog has fleas",
            "i like cold beverages",
            "the dog ate my homework",
            "don`t have a cow man",
            "i don`t think i like fleas"
    };
    
    @Override
    public void open(Map config, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector; 
    }
    
    /**
     * 轮询发数组中的某个元素
     */
    public void nextTuple() {
        this.collector.emit(new Values(sentences[index]));
        index++;
        if(index >= sentences.length){
            index = 0;
        }
        Utils.waitForSeconds(1);
    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("sentence"));
        
    }
    public void ack(Object arg0) {
    
    }
    public void activate() {
    
    }

    /**
     * 停止之前
     */
    public void close() {
        System.out.println("spout停止啦！！！！");
    }
    public void deactivate() {
    
    }
    public void fail(Object arg0) {
    
    }
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

}

【bolt1】空格切分

public class WordSplitBolt implements IRichBolt {
    
    private OutputCollector collector;
    //初始化方法
    public void prepare(Map config, TopologyContext context, OutputCollector collector) {
        this.collector = collector; 
    }
    
    public void execute(Tuple tuple) {  
        String sentence = tuple.getStringByField("sentence");
        String[] words = sentence.split(" ");
        for(String word : words){
            this.collector.emit(new Values(word));
        }
    }
    
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));
    }

    public void cleanup() {
        
    }
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}

【bolt2】汇总map

public class WordCountBolt implements IRichBolt {
    
    private OutputCollector collector;
    private HashMap<String,Long> counts = null;//单词   - 个数
    
    public void prepare(Map config, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        this.counts = new HashMap<String , Long>();
    }
    public void execute(Tuple tuple) {  
        String word = tuple.getStringByField("word");
        Long count = this.counts.get(word);
        if(count == null){
            count = 0l;
        }
        count++;
        this.counts.put(word, count);
        this.collector.emit(new Values(word,count));
    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word","count"));
    }
    public void cleanup() { 
    }
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}

【bolt3】输出结果

public class WordReportBolt implements IRichBolt {
    
    private HashMap<String,Long> counts = null;
    public void prepare(Map config, TopologyContext context, OutputCollector collector) {
        this.counts = new HashMap<String,Long>();
    }
    
    public void execute(Tuple tuple) {  
        String word = tuple.getStringByField("word");
        Long count = tuple.getLongByField("count");
        this.counts.put(word, count);
    }
    
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        
    }

    /**
     * 停止Bolt
     */
    public void cleanup() { 
        System.out.println("-----------------FINAL COUNTS----------------");
        List<String> keys = new ArrayList<String>();
        keys.addAll(this.counts.keySet());
        Collections.sort(keys);
        for(String key : keys){
            System.out.println(key + " : " + this.counts.get(key));
        }
        System.out.println("-----------------------------------------------");
    }
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}

【topology】逻辑组件

public class WordTopology {
    
    //定义常量
    private static final String WORD_SPOUT_ID = "word-spout";
    private static final String SPLIT_BOLT_ID = "split-bolt";
    private static final String COUNT_BOLT_ID = "count-bolt";
    private static final String REPORT_BOLT_ID = "report-bolt";
    private static final String TOPOLOGY_NAME = "word-count-topology";

    public static void main(String[] args) throws Exception {
        
        //实例化对象
        WordSpout spout = new WordSpout();
        WordSplitBolt splitBolt = new WordSplitBolt();
        WordCountBolt countBolt = new WordCountBolt();
        WordReportBolt reportBolt = new WordReportBolt();
        
        //构建拓扑
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout(WORD_SPOUT_ID, spout);
        
        builder.setBolt(SPLIT_BOLT_ID, splitBolt,5).shuffleGrouping(WORD_SPOUT_ID);
        
        builder.setBolt(COUNT_BOLT_ID, countBolt,5).fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));
        
        builder.setBolt(REPORT_BOLT_ID, reportBolt,10).globalGrouping(COUNT_BOLT_ID);
        
        Config config = new Config();
        config.setDebug(false);
        LocalCluster cluster = new LocalCluster();
            
        //1：本地运行
        cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());
        Utils.waitForSeconds(100);
        cluster.killTopology(TOPOLOGY_NAME);
        cluster.shutdown();
        
        //2：集群运行
        //StormSubmitter.submitTopology(TOPOLOGY_NAME, config,builder.createTopology());
    }
}

本地运行结果

2017-04-26_102746.png