MongoDB Documentation, Release 3.0.4 When operating on a sharded collection, the aggregation pipeline is split into two parts. The first pipeline runs on each shard, or if an early $match can exclude shards through the use of the shard key in the predicate, the pipeline runs on only the relevant shards. The second pipeline consists of the remaining pipeline stages and runs on the primary shard (page 669). The primary shard merges the cursors from the other shards and runs the second pipeline on these results. The primary shard forwards the final results to the mongos . In previous versions, the second pipeline would run on the mongos . 12 When splitting the aggregation pipeline into two parts, the pipeline is split to ensure that the shards perform as many stages as possible with consideration for optimization. To see how the pipeline was split, include the explain option in the db.collection.aggregate() method. Optimizations are subject to change between releases. Map-reduce supports operations on sharded collections, both as an input and as an output. This section describes the behaviors of mapReduce specific to sharded collections. When using sharded collection as the input for a map-reduce operation, mongos will automatically dispatch the map- reduce job to each shard in parallel. There is no special option required. mongos will wait for jobs on all shards to finish. Changed in version 2.2. If the out field for mapReduce has the sharded value, MongoDB shards the output collection using the _id field as the shard key. To output to a sharded collection: Optimization Map-Reduce and Sharded Collections Sharded Collection as Input Sharded Collection as Output • If the output collection does not exist, MongoDB creates and shards the collection on the _id field. • For a new or an empty sharded collection, MongoDB uses the results of the first stage of the map-reduce operation to create the initial chunks distributed among the shards. • mongos dispatches, in parallel, a map-reduce post-processing job to every shard that owns a chunk. During the post-processing, each shard will pull the results for its own chunks from the other shards, run the final reduce/finalize, and write locally to the output collection. • During later map-reduce jobs, MongoDB splits chunks as needed. • Balancing of chunks for the output collection is automatically prevented during post-processing to avoid con- currency issues. Note: 12 Until all shards upgrade to v2.6, the second pipeline runs on the mongos if any shards are still running v2.4. 7.2. Aggregation Concepts 451 In MongoDB 2.0: