[SERVER-2531] map/reduce output options need to work on sharded output collections out=(merge|reduce) Created: 10/Feb/11  Updated: 12/Jul/16  Resolved: 27/Jun/11

Status: Closed
Project: Core Server
Component/s: MapReduce
Affects Version/s: 1.7.5
Fix Version/s: 1.9.1

Type: New Feature Priority: Major - P3
Reporter: Alvin Richards (Inactive) Assignee: Antoine Girbal
Resolution: Done Votes: 6
Labels: None
Remaining Estimate: Not Specified
Time Spent: Not Specified
Original Estimate: Not Specified

Issue Links:
Depends
Related
related to SERVER-3627 sharded map-reduce output should be p... Closed
Participants:

 Description   

Right now you can't merge or reduce into a sharded collection.
For replace, we should also make sure if the old collection exists that it goes to the same place.



 Comments   
Comment by auto [ 29/Jun/11 ]

Author:

{u'login': u'agirbal', u'name': u'agirbal', u'email': u'antoine@10gen.com'}

Message: SERVER-2531: bug with finishMapReduce command, trying to create temp collection twice
Branch: master
https://github.com/mongodb/mongo/commit/7868057763ce0ba0d5b2030207588d3983d39fcf

Comment by auto [ 27/Jun/11 ]

Author:

{u'login': u'agirbal', u'name': u'agirbal', u'email': u'antoine@10gen.com'}

Message: SERVER-2531: monitor size of current key being reduced, and do extra reducing if gets too big
Branch: master
https://github.com/mongodb/mongo/commit/cbe9ebfeaf312e2a3e687e79657d0cc5a4157e60

Comment by auto [ 27/Jun/11 ]

Author:

{u'login': u'agirbal', u'name': u'agirbal', u'email': u'antoine@10gen.com'}

Message: SERVER-2531: remove possible shard chunk splitting during M/R output
Branch: master
https://github.com/mongodb/mongo/commit/17ba23539c86aa8b8c3bedc6bf6898a51ca12017

Comment by Antoine Girbal [ 27/Jun/11 ]

ganesan,
the code is mostly done, just need to do more testing and error handling.
If you want to start doing some testing and give feedback that's great.

Process is as follows, and involves only 1 mongos:

  • mongos gets the M/R command and sends it to all shards that perform their own full M/R locally into temp collection
  • upon completion, mongos does a parallel sort of results from temp collections. Since results are sorted, it can efficiently go through keys one by one and reduce duplicates.
  • the final doc for a key is inserted into a 2nd temporary collection (say temp2) that is sharded. The actual sharding used is based on the chunk info of the final collection.
  • mongos then runs the finishMapReduce command on all shards. That command will do post processing on temp2 collection, for example atomic replace or reduce into final collection.
  • a small variation for the merge mode: results are directly upserted into the final collection, no need for finishMapReduce command.

One thing I still have to figure out is how we want to handle splitting / migrating of the chunks.
Right now it assumes there is no migration of the chunks during the entire M/R process.
I think it's easier to prevent chunk migration during the M/R and let the collection get balanced when done.

I'll try to put a diagram together soon
AG

Comment by auto [ 27/Jun/11 ]

Author:

{u'login': u'agirbal', u'name': u'agirbal', u'email': u'antoine@10gen.com'}

Message: SERVER-2531: REPLACE mode now uses the finishMapReduce cmd on each shard, so that the replace is atomic.
Branch: master
https://github.com/mongodb/mongo/commit/fd3938aba48ae1f61a67c1e379bb1ea1774bf70f

Comment by ganesan pandurangan [ 27/Jun/11 ]

Hi Antoine,

Is this ticket complete, Can I use this code for testing ?

Also, on the mongos part - what is the expected behavior if there are multiple mongos ?

Will the work be shared across the multiple mongos, or will the work be done by the mongos that initially posted the map reduce command ?, Does it put any pressure on the mongos ( like memory or cpu requriements)

Can you please send me a small diagram of how things work with all your changes?

regards
ganesan

Comment by Antoine Girbal [ 27/Jun/11 ]

added REDUCE mode for sharded output.
Tried to make it optimal:

  • as with other sharded modes, mongos does parallel sort over M/R result from each shard
  • mongos reduces on the fly each key.
  • for reduce, it inserts final reduced doc to a temporary collection on each shard. Doc is inserted according to shard chunks of final collection (trick).
  • then mongos calls finishMapReduce command on each shard. The command has a special field called postProcessCollection which bypasses regular processing and just does post process.
Comment by auto [ 27/Jun/11 ]

Author:

{u'login': u'agirbal', u'name': u'agirbal', u'email': u'antoine@10gen.com'}

Message: SERVER-2531: added REDUCE mode for M/R to sharded output collection
Branch: master
https://github.com/mongodb/mongo/commit/1388ecbdb76160bbfbb2df018ed860c2c81f592a

Comment by auto [ 22/Jun/11 ]

Author:

{u'login': u'agirbal', u'name': u'agirbal', u'email': u'antoine@10gen.com'}

Message: SERVER-2531: added support for MERGE mode for M/R with sharded output
Added generic method to do update() to shards
Branch: master
https://github.com/mongodb/mongo/commit/e40135e5cef54dd71c5969c2df3966d740d6a1ea

Comment by Antoine Girbal [ 13/Jun/11 ]

It was quite a bit of work to rip out some mr code and make it executed in mongos.
But in the end it's much cleaner and efficient.
So far only the "REPLACE" mode works.

Todo:

  • monitor aggregate size of document list for current key being fetched, and reduce if needed
  • add support for MERGE mode. This would need to create a sharded update() method.
  • add support for REDUCE mode. Need to iron out the details.

INLINE mode will not be supported cause does not make sense.

Comment by auto [ 13/Jun/11 ]

Author:

{u'login': u'agirbal', u'name': u'agirbal', u'email': u'antoine@10gen.com'}

Message: SERVER-2531: added M/R output to shard collection for mode REPLACE
Branch: master
https://github.com/mongodb/mongo/commit/ec68f3c6d5e16d55b87e0981d6a5ef6819115997

Comment by Antoine Girbal [ 26/May/11 ]

ok sounds like there are several issues here:

  • output of a M/R is always a non-sharded collection
  • reduce and merge options from a sharded collection to a non-sharded one are broken

The 1st goal of this ticket is to make it possible to have sharded output.
This could be done as follows:

  • add "sharded" flag to "out" object.
  • The sharding key would always be _id (dont think there is need for other key)
  • right now the final reduce job is executed on the destination mongod. Instead it needs to be done within mongoS, so that inserts go to right places.
Comment by Jalmari Raippalinna [ 03/May/11 ]

Workaround:
db.outCollection.ensureIndex({_id: 1});

after first mapReduce seems to fix this. It seems that out collections do not have _id index automatically created.

Comment by Jalmari Raippalinna [ 03/May/11 ]

This happens with 1.8.1 when output collection is not sharded, but is used in sharded environment.

Our environment is MongoDB 1.8.1 with 2 shards and 3 replica sets each (6 servers).

Map/Reduces are done against sharded input collections and output into temporary non-existing collection with out:

{ reduce: mr_temp_collection }

Will get this:

Tue May 3 10:18:08 uncaught exception: map reduce failed:{
"assertion" : "Not an index cursor",
"assertionCode" : 9011,
"errmsg" : "db assertion failure",
"ok" : 0
}

Comment by Andy Gregorowicz [ 12/Apr/11 ]

There also seems to be an issue when output collections are not sharded, but the M/R job is run more than once.

When output is set to reduce, the first time the M/R job is run, it will succeed, but remove all indexes that were set on the collection and leave one called 0_1. Running the M/R job a second time will cause a "Not an index cursor" error.

When the output is set to merge, the first job succeeds and on the second run of the M/R job one of the shard servers will consume 100% of a CPU core and never return.

I am seeing this behavior on 1.8.1.

Generated at Thu Feb 08 03:00:18 UTC 2024 using Jira 9.7.1#970001-sha1:2222b88b221c4928ef0de3161136cc90c8356a66.