import com.mongodb.client.MapReduceIterable; //导入依赖的package包/类
private Stream<JsonObject> performMapReduce(String dbIP, String tableName, String mapRedDir)
{
  MongoClient mClient = new MongoClient(dbIP, 27017);
  MapReduceSources mrs = MapReduceSources.fromDir(mapRedDir);
  MongoDatabase database = mClient.getDatabase(tableName);
  Map<String, MapReduceIterable<Document>> mapRedMap = new HashMap<String, MapReduceIterable<Document>>();

  for (String collName : database.listCollectionNames())
  {
    MongoCollection<Document> collection = database.getCollection(collName);
    mapRedMap.put(collName, collection.mapReduce(mrs.getMapJSCode(), mrs.getReduceJSCode()));
  }

  Stream<JsonObject> result = adapter.adaptStream(mapRedMap);
  result = result.onClose(() -> { mClient.close();});

  return result;
}
 

import com.mongodb.client.MapReduceIterable; //导入依赖的package包/类
public Stream<JsonObject> adaptStream(Map<String, MapReduceIterable<Document>> mapRedMap)
{
  Stream<JsonObject> result = Stream.empty();
  JsonParser parser = new JsonParser();

  result = mapRedMap.entrySet().stream().flatMap(e ->
    StreamSupport.stream(e.getValue().spliterator(), false).map(doc ->
    {
      JsonObject jObj = (JsonObject)(parser).parse(doc.get("_id").toString());
      jObj.addProperty("_type", e.getKey());

      return jObj;
    }));

  return result;
}
 

import com.mongodb.client.MapReduceIterable; //导入依赖的package包/类
private void applyPropertiesToCursor( MongoIterable<Document> mongoIterable, QueryProperties queryProps, 
         boolean includeMetaDataSearchLimit, boolean includeSortExpr )
 {
     if( includeMetaDataSearchLimit )
     {
         Integer searchLimit = getModel().getEffectiveMDSearchLimit( queryProps ); 
         if( searchLimit > 0 )
{
             // Apply to FindIterable or MapReduceIterable
	if ( mongoIterable instanceof FindIterable )
	{
		FindIterable<Document> findIterable = (FindIterable<Document>) mongoIterable;
		findIterable.limit( searchLimit.intValue( ) );
	}
	else if ( mongoIterable instanceof MapReduceIterable )
	{
		MapReduceIterable<Document> mapReduceIterable = (MapReduceIterable<Document>) mongoIterable;
		mapReduceIterable.limit( searchLimit.intValue( ) );
	}       
}
     }
     
     applyPropertiesToCursor( mongoIterable, queryProps, includeSortExpr );
 }
 

import com.mongodb.client.MapReduceIterable; //导入依赖的package包/类
public ProfiledMapReduceIterable(String mapFunction, String reduceFunction, MapReduceIterable<TResult> mapReduce, ProfiledMongoCollection<TDocument> collection)
{
    super();
    this.mapFunction = mapFunction;
    this.reduceFunction = reduceFunction;
    this.mapReduceIterable = mapReduce;
    this.collection = collection;
}
 

import com.mongodb.client.MapReduceIterable; //导入依赖的package包/类
public MapReduceIterable<TResult> getMapReduceIterable()
{
    return mapReduceIterable;
}
 

import com.mongodb.client.MapReduceIterable; //导入依赖的package包/类
public cfData execute(cfSession _session, cfArgStructData argStruct ) throws cfmRunTimeException {
	MongoDatabase	db	= getMongoDatabase( _session, argStruct );
	
	String collection	= getNamedStringParam(argStruct, "collection", null);
	if ( collection == null )
		throwException(_session, "please specify a collection");
	
	String map	= getNamedStringParam(argStruct, "map", null );
	if ( map == null )
		throwException(_session, "please specify a map");
	
	String reduce	= getNamedStringParam(argStruct, "reduce", null );
	if ( reduce == null )
		throwException(_session, "please specify a reduce");
	
	String outputcollection	= getNamedStringParam(argStruct, "outputcollection", null );
	if ( outputcollection == null )
		throwException(_session, "please specify a outputcollection");
	
	String action		= getNamedStringParam(argStruct, "type", "replace" ).toLowerCase();
	String finalize	= getNamedStringParam(argStruct, "finalize", null );
	cfData	query		= getNamedParam(argStruct, "query", null );
	
	try{
		MapReduceIterable<Document>	mi	= db.getCollection( collection ).mapReduce( map, reduce );
		
		if ( query != null )
			mi.filter( getDocument( query ) );
		
		if ( finalize != null )
			mi.finalizeFunction( finalize );
		
		mi.collectionName( outputcollection );
		mi.action( MapReduceAction.valueOf( action ) );
		
		
		// Kick start the map reduce
		mi.first();
		
		return cfBooleanData.TRUE;

	} catch (MongoException me){
		throwException(_session, me.getMessage());
		return null;
	}
}
 

import com.mongodb.client.MapReduceIterable; //导入依赖的package包/类
<T extends IEntity, RESULT> MapReduceIterable<RESULT> mapReduce(Class<T> entity, Class<RESULT> resultClass, String mapFunction, String reduceFunction) throws Exception;