RegionalStatisticsBolt.java

public class RegionalStatisticsBolt implements IBasicBolt {
 
    private Map<String, AtomicLong> regionalCounterMapping = new HashMap<String, AtomicLong>();
 
    @Override
    public void prepare(Map map, TopologyContext topologyContext) {
        //To change body of implemented methods use File | Settings | File Templates.
    }
 
    @Override
    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
        String ip = tuple.getString(0);
        long current = 0;
 
        AtomicLong ipCounter = regionalCounterMapping.get(ip);
        if (ipCounter == null) {
            regionalCounterMapping.put(ip, new AtomicLong(0));
        }
        else {
            current = ipCounter.incrementAndGet();
        }
 
        System.out.println("ip address " + ip + " order counter " + current);
    }
 
    @Override
    public void cleanup() {
        //To change body of implemented methods use File | Settings | File Templates.
    }
 
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        //To change body of implemented methods use File | Settings | File Templates.
    }
}