Java 类com.mongodb.MapReduceCommand 实例源码

/**
 * Map reduce.
 *
 * @param mongoOperation
 *            the mongo operation
 * @param a
 *            the a
 * @param b
 *            the b
 * @param c
 *            the c
 * @param d
 *            the d
 * @throws UnknownHostException
 */
static void calcularLocalizaciones() throws UnknownHostException {

    String map = "function () { emit(this.localizacion, {count: 1}); }";
    String reduce = " function(key, values) { var result = 0; values.forEach(function(value){ result++ }); "
            + "return result; }";

    MongoClient mongoClient = new MongoClient("localhost", 27017);
    DB db = mongoClient.getDB("craulerdb");
    DBCollection ofertas = db.getCollection("ofertas");

    MapReduceCommand cmd = new MapReduceCommand(ofertas, map, reduce, null, MapReduceCommand.OutputType.INLINE,
            null);
    MapReduceOutput out = ofertas.mapReduce(cmd);

    for (DBObject o : out.results()) {
        System.out.println(o.toString());
    }
}

@Test
public void testMapReduce() throws Exception{
   MongoClient client = new MongoClient("localhost", 27017);
   DBCollection articles = client.getDB("ensim").getCollection("articles");

   // Write map function as Javascript.
   String map = "function() { " +
         "    emit('note', this.note);" +
         "}";
   // Write reduc function as Javascript.
   String reduce = "function(key, values) { " +
         "    return Array.sum(values) / values.length; " +
         "}";

   // Execute map/reduce job without finalize function and filter query.
   MapReduceOutput out = articles.mapReduce(map, reduce, null, MapReduceCommand.OutputType.INLINE, null);

   try{
      for (DBObject o : out.results()){
         System.err.println(o.toString());
      }
   } catch (Exception e){
      e.printStackTrace();
   }
}

/**
 * runs a map-reduce-job on the collection. The functions are read from the classpath in the folder mongodb. The systems reads them from
 * files called <name>.map.js, <name>.reduce.js and optionally <name>.finalize.js. After this the result is converted
 * using the given {@link MapReduceResultHandler}
 * 
 * @param <R> the type of the result class
 * @param name the name of the map-reduce functions
 * @param query the query to filter the elements used for the map-reduce
 * @param sort sort query to sort elements before running map-reduce
 * @param scope the global scope for the JavaScript run
 * @param conv the converter to convert the result
 * @return an {@link Iterable} with the result entries
 * @throws RuntimeException if resources cannot be read
 */
protected final <R> Iterable<R> mapReduce(String name, DBObject query, DBObject sort, Map<String, Object> scope, final MapReduceResultHandler<R> conv) {
    String map = this.getMRFunction(name, "map");
    String reduce = this.getMRFunction(name, "reduce");

    MapReduceCommand mrc = new MapReduceCommand(this.collection.getDBCollection(), map, reduce, null, OutputType.INLINE, query);
    String finalizeFunction = this.getMRFunction(name, "finalize");
    if (finalizeFunction != null) {
        mrc.setFinalize(finalizeFunction);
    }
    if (sort != null) {
        mrc.setSort(sort);
    }
    if (scope != null) {
        mrc.setScope(scope);
    }
    MapReduceOutput mr = this.collection.getDBCollection().mapReduce(mrc);
    return new ConverterIterable<R>(mr.results().iterator(), conv);
}

@SuppressWarnings("deprecation")
MapReduceCommand toCommand(final Mapper mapper) {
    if (query.getOffset() != 0 || query.getFieldsObject() != null) {
        throw new QueryException("mapReduce does not allow the offset/retrievedFields query ");
    }

    final DBCollection dbColl = inputCollection != null ? getQuery().getCollection().getDB().getCollection(inputCollection)
                                                        : query.getCollection();
    final String target = outputCollection != null ? outputCollection : mapper.getMappedClass(resultType).getCollectionName();

    final MapReduceCommand command = new MapReduceCommand(dbColl, map, reduce, target, outputType, query.getQueryObject());
    command.setBypassDocumentValidation(bypassDocumentValidation);
    command.setCollation(collation);
    command.setFinalize(finalize);
    command.setJsMode(jsMode);
    command.setLimit(limit);
    command.setMaxTime(maxTimeMS, TimeUnit.MILLISECONDS);
    command.setOutputDB(outputDB);
    command.setReadPreference(readPreference);
    command.setScope(scope);
    command.setSort(query.getSortObject());
    command.setVerbose(verbose);

    return command;
}

public ContributorsCounter(String host, String dbname, String collectionName, String output) throws Exception {
    MongoClient mongo = null;
    try {
        mongo = new MongoClient(host, 27017);
    } catch (UnknownHostException e) {
        throw new Exception(e);
    }
    DB db = mongo.getDB(dbname);
    this.collection = db.getCollection(collectionName);

    this.outputType = MapReduceCommand.OutputType.REPLACE;

    DBObject query = new BasicDBObject();
    query.put("author", new BasicDBObject("$exists", Boolean.TRUE));

    //this.output = output;
    this.mr_cmd = new MapReduceCommand(collection, map, reduce, output, outputType, query);

}

public DomainsCounter(String host, String dbname, String collectionName, String output) throws Exception {
    MongoClient mongo = null;
    try {
        mongo = new MongoClient(host, 27017);
    } catch (UnknownHostException e) {
        throw new Exception(e);
    }
    DB db = mongo.getDB(dbname);
    this.collection = db.getCollection(collectionName);

    this.outputType = MapReduceCommand.OutputType.REPLACE;

    DBObject query = new BasicDBObject();
    query.put("url", new BasicDBObject("$exists", Boolean.TRUE));

    //this.output = output;
    this.mr_cmd = new MapReduceCommand(collection, map, reduce, output, outputType, query);

}

public TagsCounter(String host, String dbname, String collectionName, String output) throws Exception {
    MongoClient mongo = null;
    try {
        mongo = new MongoClient(host, 27017);
    } catch (UnknownHostException e) {
        throw new Exception(e);
    }
    DB db = mongo.getDB(dbname);
    this.collection = db.getCollection(collectionName);

    this.outputType = MapReduceCommand.OutputType.REPLACE;

    DBObject query = new BasicDBObject();
    query.put("tags", new BasicDBObject("$ne", new String[0]));

    //this.output = output;
    this.mr_cmd = new MapReduceCommand(collection, map, reduce, output, outputType, query);

}

public ContributorsCounter(String host, String dbname, String collectionName, String output) throws Exception {
    MongoClient mongo = null;
    try {
        mongo = new MongoClient(host, 27017);
    } catch (UnknownHostException e) {
        throw new Exception(e);
    }
    DB db = mongo.getDB(dbname);
    this.collection = db.getCollection(collectionName);

    this.outputType = MapReduceCommand.OutputType.REPLACE;

    DBObject query = new BasicDBObject();
    query.put("uid", new BasicDBObject("$exists", Boolean.TRUE));

    //this.output = output;
    this.mr_cmd = new MapReduceCommand(collection, map, reduce, output, outputType, query);

}

public DomainsCounter(String host, String dbname, String collectionName, String output) throws Exception {
    MongoClient mongo = null;
    try {
        mongo = new MongoClient(host, 27017);
    } catch (UnknownHostException e) {
        throw new Exception(e);
    }
    DB db = mongo.getDB(dbname);
    this.collection = db.getCollection(collectionName);

    this.outputType = MapReduceCommand.OutputType.REPLACE;

    DBObject query = new BasicDBObject();
    query.put("url", new BasicDBObject("$exists", Boolean.TRUE));

    //this.output = output;
    this.mr_cmd = new MapReduceCommand(collection, map, reduce, output, outputType, query);

}

public TagsCounter(String host, String dbname, String collectionName, String output) throws Exception {
    MongoClient mongo = null;
    try {
        mongo = new MongoClient(host, 27017);
    } catch (UnknownHostException e) {
        throw new Exception(e);
    }
    DB db = mongo.getDB(dbname);
    this.collection = db.getCollection(collectionName);

    this.outputType = MapReduceCommand.OutputType.REPLACE;

    DBObject query = new BasicDBObject();
    query.put("tags", new BasicDBObject("$ne", new String[0]));

    //this.output = output;
    this.mr_cmd = new MapReduceCommand(collection, map, reduce, output, outputType, query);

}

private synchronized Iterable<DBObject> mapReduce(String map, String reduce, String outputTarget, MapReduceCommand.OutputType outputType, String orderBy, DBObject query) {
    MapReduceOutput output = getCollection().mapReduce(map, reduce, outputTarget, outputType, query);
    DBCollection c = output.getOutputCollection();
    DBCursor cursor;
    if(orderBy != null){
        cursor = c.find().sort(SortUtil.getSort(orderBy));
    }else{
        cursor = c.find();
    }
    List<DBObject> list = new ArrayList<DBObject>();
    for(Iterator<DBObject> it = cursor.iterator(); it.hasNext(); ){
        list.add(it.next());
    }
    return list;
}

private synchronized Iterable<DBObject> mapReduce(String map, String reduce, String outputTarget, MapReduceCommand.OutputType outputType, String orderBy, int pageNum, int pageSize, DBObject query) {
    MapReduceOutput output = getCollection().mapReduce(map, reduce, outputTarget, outputType, query);
    DBCollection c = output.getOutputCollection();
    DBCursor cursor;
    if(orderBy != null){
        cursor = c.find().sort(SortUtil.getSort(orderBy)).skip((pageNum-1)*pageSize).limit(pageSize);
    }else{
        cursor = c.find().skip((pageNum-1)*pageSize).limit(pageSize);
    }
    List<DBObject> list = new ArrayList<DBObject>();
    for(Iterator<DBObject> it = cursor.iterator(); it.hasNext(); ){
        list.add(it.next());
    }
    return list;
}

@Override
@Deprecated
public <T> MapreduceResults<T> mapReduce(final MapreduceType type, final Query query, final String map, final String reduce,
                                         final String finalize, final Map<String, Object> scopeFields, final Class<T> outputType) {

    final DBCollection dbColl = query.getCollection();

    final String outColl = mapper.getCollectionName(outputType);

    final MapReduceCommand cmd = new MapReduceCommand(dbColl, map, reduce, outColl, type.toOutputType(), query.getQueryObject());

    if (query.getLimit() > 0) {
        cmd.setLimit(query.getLimit());
    }
    if (query.getSortObject() != null) {
        cmd.setSort(query.getSortObject());
    }

    if (finalize != null && finalize.length() != 0) {
        cmd.setFinalize(finalize);
    }

    if (scopeFields != null && !scopeFields.isEmpty()) {
        cmd.setScope(scopeFields);
    }

    return mapReduce(type, query, outputType, cmd);
}

@GET
@Path("movies/{id}")
@Produces(MediaType.APPLICATION_JSON)
public static String getMovies(@PathParam("id") String userId) {
    Logger.getLogger(XPostWS.class.getName()).log(Level.INFO, "retrieving movies for " + userId);

    final JsonObject result = new JsonObject();

    Mongo mongo = getActionMongo();
    final DB db = mongo.getDB("mydb");
    final DBCollection moviesCol = db.getCollection("movies");

    String map = "function () {" + "if (this.userId != this.friendId)" + "{emit(this.id, this);}" + "};";
    String reduce = "function (key, values) {" +
                        "var res = {};" +
                        "res.name = values[0].name;" +
                        "res.id = key;" +
                        "res.userId = values[0].userId;" +
                        "res.poster = values[0].poster;" +
                        "res.rating = values[0].rating;" +
                        "res.imdbId = values[0].imdbId;" +
                        "res.friendId = [];" +
                        "for (var i = 0; i<values.length; ++i) {" +
                            "res.friendId.push(String(values[i].friendId));" +
                        "}" +
                        "return res;}";

    final BasicDBObject searchObject = new BasicDBObject("userId", userId);
    MapReduceCommand mapReduceComand = new MapReduceCommand(moviesCol, map, reduce, null,
            MapReduceCommand.OutputType.INLINE, searchObject);
    final MapReduceOutput moviesReduced = moviesCol.mapReduce(mapReduceComand);
    JsonArray moviesInfo = new JsonArray();
    for (DBObject movie : moviesReduced.results()) {
        moviesInfo.put(new JsonObject(movie.toString()));
    }
    Logger.getLogger(XPostWS.class.getName()).log(Level.INFO,
            "retrieved " + moviesInfo.length() + " movies for user " + userId);
    if (moviesInfo.length() == 0) {
        result.put("movies", new JsonArray().toString());
    } else {
        sortMovies(moviesInfo, userId);
        result.put("movies", moviesInfo);
    }

    try {
        return new String(result.toString().getBytes("UTF-8"));
    } catch (UnsupportedEncodingException e) {
        Logger.getLogger(TAG).severe(e.getMessage());
        return result.toString();
    }
}

public List<JSONObject> getComments(String word) throws UnknownHostException, MongoException, SQLException, ClassNotFoundException, JSONException {
    // TODO Auto-generated method stub
    List<JSONObject> listeComments = new ArrayList<JSONObject>();

    Mongo mongoClient = new Mongo();
    DB db = mongoClient.getDB("social");
    DBCollection coll = db.getCollection("comments");   

    String m="function wordMap(){"+
            "var text = this.comment;"+
            "var words = text.match(/\\w+/g);"+
            "var tf = new Array();"+
            "for( var i=0; i< words.length ; i++ ){"+
            "if( tf[words[i]] == null){"+
            "tf[words[i]]=1;"+
            "}"+
            "else{"+    
            "tf[words[i]]++;"+
            "}"+
            "}"+
            "for( var i=0; i< words.length ; i++ ){"+
            "emit(this._id, { word : words[i], tf : tf[words[i]] } )}};";

    String r="function wordReduce(key, values){"+
            "return ( { \"tfs\" : values } )};";

    MapReduceOutput out = coll.mapReduce(m,r,null,MapReduceCommand.OutputType.INLINE,null);

    String[] tabWords = word.split(" ");
    SortedMap<String,String> map = new TreeMap<String,String>();


    for ( DBObject obj : out.results()){

        JSONObject jsonObj = new JSONObject(obj.toMap());


        String idObj = jsonObj.get("_id").toString();
        JSONObject jsonValue = new JSONObject(jsonObj.get("value").toString());
        JSONArray tfsTab = new JSONArray(jsonValue.get("tfs").toString());


        for(int i=0; i<tfsTab.length();i++){
            JSONObject tfsObj = tfsTab.getJSONObject(i);
            for(String w : tabWords){
                if(tfsObj.get("word").equals(w)){
                    String nbtf = tfsObj.get("tf").toString();
                    map.put(idObj,nbtf);
                }
            }
        }



    }

    mongoClient.close();


    Iterator iterator = map.keySet().iterator();
    while (iterator.hasNext()) {
        Object key = iterator.next();
        JSONObject objToAdd = new JSONObject();
        objToAdd.put("_id",key.toString());
        listeComments.add(objToAdd);
    }


    return listeComments;
}

public Iterable<DBObject> mapReduce(MapReduceCommand cmd) {
    MapReduceOutput output = getCollection().mapReduce(cmd);
    return output.results();
}

public Iterable<DBObject> mapReduce(String map, String reduce) {
    MapReduceOutput output = getCollection().mapReduce(map, reduce, null, MapReduceCommand.OutputType.INLINE, null);
    return output.results();
}

private Iterable<DBObject> mapReduce(String map, String reduce, DBObject query) {
    MapReduceOutput output = getCollection().mapReduce(map, reduce, null, MapReduceCommand.OutputType.INLINE, query);
    return output.results();
}

public Iterable<DBObject> mapReduce(String map, String reduce, String outputTarget, MapReduceCommand.OutputType outputType, String orderBy, BuguQuery query) {
    return mapReduce(map, reduce, outputTarget, outputType, orderBy, query.getCondition());
}

public Iterable<DBObject> mapReduce(String map, String reduce, String outputTarget, MapReduceCommand.OutputType outputType, String orderBy, int pageNum, int pageSize, BuguQuery query) {
    return mapReduce(map, reduce, outputTarget, outputType, orderBy, pageNum, pageSize, query.getCondition());
}

@Override
@Deprecated
public <T> MapreduceResults<T> mapReduce(final MapreduceType type, final Query query, final Class<T> outputType,
                                         final MapReduceCommand baseCommand) {

    Assert.parametersNotNull("map", baseCommand.getMap());
    Assert.parameterNotEmpty("map", baseCommand.getMap());
    Assert.parametersNotNull("reduce", baseCommand.getReduce());
    Assert.parameterNotEmpty("reduce", baseCommand.getReduce());

    if (query.getOffset() != 0 || query.getFieldsObject() != null) {
        throw new QueryException("mapReduce does not allow the offset/retrievedFields query options.");
    }

    final OutputType outType = type.toOutputType();

    final DBCollection dbColl = query.getCollection();

    final MapReduceCommand cmd = new MapReduceCommand(dbColl, baseCommand.getMap(), baseCommand.getReduce(),
                                                      baseCommand.getOutputTarget(), outType, query.getQueryObject());
    cmd.setFinalize(baseCommand.getFinalize());
    cmd.setScope(baseCommand.getScope());

    if (query.getLimit() > 0) {
        cmd.setLimit(query.getLimit());
    }
    if (query.getSortObject() != null) {
        cmd.setSort(query.getSortObject());
    }

    if (LOG.isTraceEnabled()) {
        LOG.info("Executing " + cmd.toString());
    }

    final EntityCache cache = createCache();
    MapreduceResults<T> results = new MapreduceResults<T>(dbColl.mapReduce(baseCommand));

    results.setType(type);
    if (MapreduceType.INLINE.equals(type)) {
        results.setInlineRequiredOptions(this, outputType, getMapper(), cache);
    } else {
        results.setQuery(newQuery(outputType, getDB().getCollection(results.getOutputCollectionName())));
    }

    return results;

}

@Test
@SuppressWarnings("deprecation")
public void mapReduceCommand() {
    Query<FacebookUser> query = getDs().find(FacebookUser.class);
    MapReduceOptions<FacebookUser> options = new MapReduceOptions<FacebookUser>()
        .bypassDocumentValidation(true)
        .collation(Collation.builder().locale("en").build())
        .finalize("i'm a finalize function")
        .jsMode(true)
        .limit(42)
        .map("i'm a map function")
        .maxTimeMS(42000)
        .outputCollection("output collection")
        .outputDB("output db")
        .outputType(OutputType.INLINE)
        .query(query)
        .readPreference(ReadPreference.primaryPreferred())
        .reduce("i'm a reduce function")
        .scope(new Document("key", "value").append("key2", "value2"))
        .verbose(true);

    MapReduceCommand command = options.toCommand(getMorphia().getMapper());

    assertTrue(command.getBypassDocumentValidation());
    assertEquals(Collation.builder().locale("en").build(), command.getCollation());
    assertTrue(command.getJsMode());
    assertEquals(42, command.getLimit());
    assertEquals("i'm a map function", command.getMap());
    assertEquals(42000, command.getMaxTime(TimeUnit.MILLISECONDS));
    assertEquals("output collection", command.getOutputTarget());
    assertEquals("output db", command.getOutputDB());
    assertEquals(query.getQueryObject(), command.getQuery());
    assertEquals(query.getSortObject(), command.getSort());
    assertEquals(ReadPreference.primaryPreferred(), command.getReadPreference());
    assertEquals("i'm a map function", command.getMap());
    assertEquals("i'm a reduce function", command.getReduce());
    assertEquals("i'm a finalize function", command.getFinalize());
    assertEquals(new Document("key", "value").append("key2", "value2"), command.getScope());
    assertTrue(command.isVerbose());

}

/**
 * Runs a map/reduce job at the server; this should be used with a server version 1.7.4 or higher
 *
 * @param <T>         The type of resulting data
 * @param type        MapreduceType
 * @param q           The query (only the criteria, limit and sort will be used)
 * @param outputType  The type of resulting data; inline is not working yet
 * @param baseCommand The base command to fill in and send to the server
 * @return counts and stuff
 * @deprecated use {@link #mapReduce(MapReduceOptions)} instead
 */
@Deprecated
<T> MapreduceResults<T> mapReduce(MapreduceType type, Query q, Class<T> outputType, MapReduceCommand baseCommand);