TridentWordCount

package test;

import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.FilterNull;
import storm.trident.operation.builtin.MapGet;
import storm.trident.operation.builtin.Sum;
import storm.trident.testing.FixedBatchSpout;
import storm.trident.testing.MemoryMapState;
import storm.trident.testing.Split;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;


public class TridentWordCount {

    public static StormTopology buildTopology(LocalDRPC drpc) {

        // Spout para el test
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("title", "text"), 3,
                new Values("Don Quijote de la Mancha", "Es una novela escrita por el español Miguel de Cervantes Saavedra. Publicada su primera parte con el título de El ingenioso hidalgo don Quijote de la Mancha a comienzos de 1605"),
                new Values("El nombre de la rosa", "Es una novela de misterio e histórica de Umberto Eco publicada en 1980"),
                new Values("Lazarillo de Tormes", "Es una novela española anónima, escrita en primera persona y en estilo epistolar (como una sola y larga carta), cuya edición conocida más antigua data de 1554"));

        // Topologia para el contado de palabras
        TridentTopology topology = new TridentTopology();
        TridentState wordCounts =
              topology.newStream("spout", spout)
                .parallelismHint(3)
                .each(new Fields("text"), new Split(), new Fields("word"))
                .groupBy(new Fields("word"))
                .persistentAggregate(new MemoryMapState.Factory(),
                                     new Count(), new Fields("count"))
                .parallelismHint(3);

        // Topologia DRPC para la consulta del numero de ocurrencias de las palabras
        topology.newDRPCStream("words", drpc)
                .each(new Fields("args"), new Split(), new Fields("word"))
                .groupBy(new Fields("word"))
                .stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count"))
                .each(new Fields("count"), new FilterNull())
                .aggregate(new Fields("count"), new Sum(), new Fields("sum"));

        return topology.build();
    }

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        conf.setMaxSpoutPending(20);
        LocalDRPC drpc = new LocalDRPC();
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("wordCounter", conf, buildTopology(drpc));
        for(int i=0; i<2; i++) {
            System.out.println("Nº de ocurrencias de Es: " + drpc.execute("words", "Es"));
            System.out.println("Nº de ocurrencias de una: " + drpc.execute("words", "una"));
            System.out.println("Nº de ocurrencias de novela: " + drpc.execute("words", "novela"));
            System.out.println("Nº de ocurrencias de primera:" + drpc.execute("words", "primera"));
            Thread.sleep(1000);
        }
    }
}