WordCountExample.java

import org.infinispan.Cache;
import org.infinispan.distexec.mapreduce.Collector;
import org.infinispan.distexec.mapreduce.MapReduceTask;
import org.infinispan.distexec.mapreduce.Mapper;
import org.infinispan.distexec.mapreduce.Reducer;
import org.infinispan.manager.DefaultCacheManager;

import java.util.Iterator;
import java.util.Map;
import java.util.StringTokenizer;

/**
 * Date: 2/14/13
 * Time: 9:08 AM
 */


public class WordCountExample {

    /**
     * In this example replace c1 and c2 with
     * real Cache references
     *
     * @param args
     */
    public static void main(String[] args) {

        DefaultCacheManager m = new DefaultCacheManager();

        Cache<String,String> c1 = m.getCache();
        Cache<String,String> c2 = m.getCache();

        c1.put("1", "Hello world here I am");
        c2.put("2", "Infinispan rules the world");
        c1.put("3", "JUDCon is in Boston");
        c2.put("4", "JBoss World is in Boston as well");
        c1.put("12","JBoss Application Server");
        c2.put("15", "Hello world");
        c1.put("14", "Infinispan community");
        c2.put("15", "Hello world");

        c1.put("111", "Infinispan open source");
        c2.put("112", "Boston is close to Toronto");
        c1.put("113", "Toronto is a capital of Ontario");
        c2.put("114", "JUDCon is cool");
        c1.put("211", "JBoss World is awesome");
        c2.put("212", "JBoss rules");
        c1.put("213", "JBoss division of RedHat ");
        c2.put("214", "RedHat community");

        MapReduceTask<String, String, String, Integer> t =
                new MapReduceTask<String, String, String, Integer>(c1);
        t.mappedWith(new WordCountMapper())
                .reducedWith(new WordCountReducer());
        Map<String, Integer> wordCountMap = t.execute();
    }

    static class WordCountMapper implements Mapper<String,String,String,Integer> {
        /** The serialVersionUID */
        private static final long serialVersionUID = -5943370243108735560L;

        @Override
        public void map(String key, String value, Collector<String, Integer> c) {
            StringTokenizer tokens = new StringTokenizer(value);
            while (tokens.hasMoreElements()) {
                String s = (String) tokens.nextElement();
                c.emit(s, 1);
            }
        }
    }

    static class WordCountReducer implements Reducer<String, Integer> {
        /** The serialVersionUID */
        private static final long serialVersionUID = 1901016598354633256L;

        @Override
        public Integer reduce(String key, Iterator<Integer> iter) {
            int sum = 0;
            while (iter.hasNext()) {
                Integer i = (Integer) iter.next();
                sum += i;
            }
            return sum;
        }
    }
}