MongoDB Incremental Map Reduce script to aggregate unique values in batches based on date

mongo_mapreduce.js

map = function() {
	if (!this.fruits) return;
	var skip_fruits = {
		'Watermelon':1,
		'Grapefruit':1,
		'Tomato':1 // yes, a tomato is a fruit
	}
	for (var fruit in this.fruits) {
		if (skip_fruits[fruit]) continue;
		var obj = {};
		obj[this.fruits[fruit]] = 1;
		emit(fruit, obj);
	}
};

reduce = function(key, values) {
	var out_values = {};
	values.forEach(function(v) {
		for(var k in v) { // iterate values
			if (!out_values[k]) {
				out_values[k] = v[k]; // init missing counter
			} else {
				out_values[k] += v[k];
			}
		}
	});
	return out_values;
};

var in_coll = "repo";
var out_coll = "fruit_agg_so";
var total_docs = db[in_coll].count();
var page_size = 50000;
var pages = Math.floor(total_docs / page_size);
print('Starting incremental MR job with '+pages+' pages');
db[out_coll].drop();
for (var i=0; i<pages; i++) {
	var skip = page_size * i;
	print("Calculating page limits for "+skip+" - "+(skip+page_size-1)+"...");
	var start_date = db[in_coll].find({},{date:1}).sort({date:1}).skip(skip).limit(1)[0].date;
	var end_date = db[in_coll].find({},{date:1}).sort({date:1}).skip(skip+page_size-1).limit(1)[0].date;
	var mr_command = {
		mapreduce: in_coll,
		map: map,
		reduce: reduce,
		out: {reduce: out_coll},
		sort: {date: 1},
		query: {
			date: {
				$gte: start_date,
				$lt: end_date
			}
		},
		limit: (page_size - 1)
	};
	print("Running mapreduce for "+skip+" - "+(skip+page_size-1));
	db[in_coll].runCommand(mr_command);
}

// Show number of number of possible values per key
var keys = [];
for (var c = db.fruit_agg_so.find(); c.hasNext();) {
	var obj = c.next();
	if (!obj.value) break;
	var len=0;for(var l in obj.value){len++;}
	keys.push({key: obj['_id'], value: len});
}
keys.sort(function(a, b){
	if (a.value == b.value) return 0;
	return (a.value > b.value)? -1: 1;
});
for (var i=0; i<20; i++) {
	print(keys[i].key+':'+keys[i].value);
}