1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
// mr.js
MR = {};
MR.init = function(){
$max = 0;
$arr = [];
emit = MR.emit;
$numEmits = 0;
$numReduces = 0;
$numReducesToDB = 0;
gc(); // this is just so that keep memory size sane
}
MR.cleanup = function(){
MR.init();
gc();
}
MR.emit = function(k,v){
$numEmits++;
var num = nativeHelper.apply( get_num_ , [ k ] );
var data = $arr[num];
if ( ! data ){
data = { key : k , values : new Array(1000) , count : 0 };
$arr[num] = data;
}
data.values[data.count++] = v;
$max = Math.max( $max , data.count );
}
MR.doReduce = function( useDB ){
$numReduces++;
if ( useDB )
$numReducesToDB++;
$max = 0;
for ( var i=0; i<$arr.length; i++){
var data = $arr[i];
if ( ! data )
continue;
if ( useDB ){
var x = tempcoll.findOne( { _id : data.key } );
if ( x ){
data.values[data.count++] = x.value;
}
}
var r = $reduce( data.key , data.values.slice( 0 , data.count ) );
if ( r && r.length && r[0] ){
data.values = r;
data.count = r.length;
}
else{
data.values[0] = r;
data.count = 1;
}
$max = Math.max( $max , data.count );
if ( useDB ){
if ( data.count == 1 ){
tempcoll.save( { _id : data.key , value : data.values[0] } );
}
else {
tempcoll.save( { _id : data.key , value : data.values.slice( 0 , data.count ) } );
}
}
}
}
MR.check = function(){
if ( $max < 2000 && $arr.length < 1000 ){
return 0;
}
MR.doReduce();
if ( $max < 2000 && $arr.length < 1000 ){
return 1;
}
MR.doReduce( true );
$arr = [];
$max = 0;
reset_num();
gc();
return 2;
}
MR.finalize = function(){
tempcoll.find().forEach(
function(z){
z.value = $finalize( z._id , z.value );
tempcoll.save( z );
}
);
}
|
