Apache Word Count Code

Apache Word Count Code-->

This is basic word count Storm code. copy paste this code to eclipse in different java files and run the main method including jars for storm which can be downloaded from Apache Storm website.

1. WordCountTopology .java --> main topology class.

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;


public class WordCountTopology {

    private static final String SENTENCE_SPOUT_ID = "sentence-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 {

        SentenceSpout spout = new SentenceSpout();
        SplitSentenceBolt splitBolt = new SplitSentenceBolt();
        countBolt countBolt = new countBolt();
        ReportBolt reportBolt = new ReportBolt();


        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout(SENTENCE_SPOUT_ID, spout);
        // SentenceSpout --> SplitSentenceBolt
        builder.setBolt(SPLIT_BOLT_ID, splitBolt)
                .shuffleGrouping(SENTENCE_SPOUT_ID);
        // SplitSentenceBolt --> WordCountBolt
        builder.setBolt(COUNT_BOLT_ID, countBolt)
               .fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));
        // WordCountBolt --> ReportBolt
        builder.setBolt(REPORT_BOLT_ID, reportBolt)
                .globalGrouping(COUNT_BOLT_ID);

        Config config = new Config();

        LocalCluster cluster = new LocalCluster();

        cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());
        Utils.sleep(10000);
        cluster.killTopology(TOPOLOGY_NAME);
        cluster.shutdown();
    }
}

 2. SentanceSpout

import backtype.storm.spout.SpoutOutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.base.BaseRichSpout;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Values;

import backtype.storm.utils.Utils;

import java.util.Map;

public class SentenceSpout extends BaseRichSpout {

    private SpoutOutputCollector collector;

    private String[] sentences = {

        "I love big data",

        "I also love real time analytics",

        "i love hadoop",

        "i love kafka",

        "i love spark"

    };

    private int index = 0;

    public void declareOutputFields(OutputFieldsDeclarer declarer) {

        declarer.declare(new Fields("sentence"));

    }

    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;

        }

        

    }

}

3 .   SplitSentenceBolt .java

import backtype.storm.task.OutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.base.BaseRichBolt;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Tuple;

import backtype.storm.tuple.Values;

import java.util.Map;

public class SplitSentenceBolt extends BaseRichBolt{

    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"));

    }

}

4 CountBolt.java

import java.util.HashMap;

import java.util.Map;

import backtype.storm.task.OutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.base.BaseRichBolt;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Tuple;

import backtype.storm.tuple.Values;

public class countBolt extends BaseRichBolt{

private OutputCollector  collector;

private HashMap<String,Long> map = new HashMap<String,Long>();

@Override

public void execute(Tuple tuple) {

String word  = tuple.getStringByField("word");

Long count = map.get(word);

if(count == null)

{

count=1L;

}else{

count++;

}

map.put(word,count);

//System.out.println(word+":"+count);

collector.emit(new Values(word,count));

}

@Override

public void prepare(Map map, TopologyContext context, OutputCollector collector) {

this.collector = collector;

}

@Override

public void declareOutputFields(OutputFieldsDeclarer declarer) {

declarer.declare(new Fields("word","count"));

}

}

5. ReportBolt.java

import java.util.Map;

import backtype.storm.task.OutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.base.BaseRichBolt;

import backtype.storm.tuple.Tuple;

public class ReportBolt extends BaseRichBolt{

@Override

public void execute(Tuple tuple) {

String word = tuple.getStringByField("word");

Long count = tuple.getLongByField("count");

System.out.println("word-->"+ word);

System.out.println("count-->"+ count);

}

@Override

public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {

}

@Override

public void declareOutputFields(OutputFieldsDeclarer arg0) {

}

}