MongoDB:MapReduce基礎及び実例

背景
     MapReduceは非常に柔軟で強力なデータ統合ツールである.その利点は、一つの統合タスクを複数の小さなタスクに分解し、複数のサーバに割り当てて並列処理することです.
MongoDBもMapReduceを提供しています.もちろん、検索語はJavaScriptです.MongoDBのMapReduceは主に以下の段階があります.
   1.Map:Mapをセット内の各ドキュメントに操作します.
   2.Shuffle:Keyグループに従ってドキュメントに対して、異なるKey毎に一連の(>=1個)の値表(List of values)を生成する.
   3.Reduce:値テーブルの要素が一つしかないまで処理します.その後、値テーブルをShuffleプロセスに戻し、各Keyが一つの値テーブルにのみ対応するまで循環処理し、この値テーブルには一つの要素しかない.これはMRの結果である.
   4.Finalize:このステップは必須ではありません.MRの最終結果を得た後、いくつかのデータ「トリミング」特性の処理を行う.
MongoDBではeit関数を使ってMapReduceにKey/Valueペアを提供します.
Reduce関数は二つのパラメータを受け取ります.Key,emits.Keyはemit関数のKeyです.emitesは配列で、その要素はemit関数が提供するValueです.
Reduce関数の戻り結果はMapまたはReduceで重複して使用される必要がありますので、戻り結果はemitsの要素構造と一致しなければなりません.
MapまたはReduce関数の中のthisキーワードは、現在のMapping文書を表します.
実例
  テストデータ:このセットは3人のユーザーが購入した製品と製品の価格のデータです.

for(var i=0;i<1000;i++){ 
var rID=Math.floor(Math.random()*10); 
//Math.random()*10, 0-9   ，Math.floor(Math.random()*10)     
var price=parseFloat((Math.random()*10).toFixed(2)); 
//(Math.random()*10).toFixed(2)    ，parseFloat()    
if(rID<4){ 
    db.test.insert({"user":"Joe","sku":rID,"price":price}); 
} 
else if(rID>=4 && rID<7)
{ 
   db.test.insert({"user":"Josh","sku":rID,"price":price}); 
} 
else {  
db.test.insert({"user":"Ken","sku":rID,"price":price}); 
     } 
}

1.各ユーザーはそれぞれいくつの製品を買いましたか?(

//SQL  select user,count(sku) from test
group by user//MapReduce  map=function (){
 emit(this.user,{count:1})
}

reduce=function (key,values){ var cnt=0;   
values.forEach(function(val){ cnt+=val.count;});  
return {"count":cnt};
}//MR      mr1db.test.mapReduce(map,reduce,{out:"mr1"})//  MR    > db.mr1.find()
{ "_id" : "Joe", "value" : { "count" : 416 } }
{ "_id" : "Josh", "value" : { "count" : 287 } }
{ "_id" : "Ken", "value" : { "count" : 297 } }

2.ユーザーごとに異なる製品はいくつ買いましたか?(複合KeyはMRをします)

//SQL  select user,sku,count(*) from test
group by user,sku//MapReduce  map=function (){
 emit({user:this.user,sku:this.sku},{count:1})
}

reduce=function (key,values){ var cnt=0;   
values.forEach(function(val){ cnt+=val.count;});  
return {"count":cnt};
}

db.test.mapReduce(map,reduce,{out:"mr2"})
> db.mr2.find()
{ "_id" : { "user" : "Joe", "sku" : 0 }, "value" : { "count" : 103 } }
{ "_id" : { "user" : "Joe", "sku" : 1 }, "value" : { "count" : 106 } }
{ "_id" : { "user" : "Joe", "sku" : 2 }, "value" : { "count" : 102 } }
{ "_id" : { "user" : "Joe", "sku" : 3 }, "value" : { "count" : 105 } }
{ "_id" : { "user" : "Josh", "sku" : 4 }, "value" : { "count" : 87 } }
{ "_id" : { "user" : "Josh", "sku" : 5 }, "value" : { "count" : 107 } }
{ "_id" : { "user" : "Josh", "sku" : 6 }, "value" : { "count" : 93 } }
{ "_id" : { "user" : "Ken", "sku" : 7 }, "value" : { "count" : 98 } }
{ "_id" : { "user" : "Ken", "sku" : 8 }, "value" : { "count" : 83 } }
{ "_id" : { "user" : "Ken", "sku" : 9 }, "value" : { "count" : 116 } }

3.ユーザーごとに購入した製品の数は、総金額はいくらですか?(複合Reduce結果処理)

//SQL  select user,count(sku),sum(price) from test
group by user//MapReduce  map=function (){
  emit(this.user,{amount:this.price,count:1})
}

reduce=function (key,values){ var res={amount:0,count:0}
values.forEach(function(val){ 
    res.amount+=val.amount;
    res.count+=val.count
});  
return res;
}

db.test.mapReduce(map,reduce,{out:"mr3"})

> db.mr3.find()
{ "_id" : "Joe", "value" : { "amount" : 2053.8899999999994, "count" : 395 } }
{ "_id" : "Josh", "value" : { "amount" : 1409.2600000000002, "count" : 292 } }
{ "_id" : "Ken", "value" : { "amount" : 1547.7700000000002, "count" : 313 } }

4.3で返したamountのfloat精度を2桁の小数に変更する必要があり、商品の平均価格を得る必要があります.(Finalizeを使ってreduce結果集を処理します.)

//SQL  select user,cast(sum(price) as   decimal(10,   2)) as amount,count(sku) as [count],
cast((sum(price)/count(sku))  as decimal(10,2)) as avgPrice
from test
group by user//MapReduce  map=function (){
  emit(this.user,{amount:this.price,count:1,avgPrice:0})
}

reduce=function (key,values){ var res={amount:0,count:0,avgPrice:0}
values.forEach(function(val){ 
    res.amount+=val.amount;
    res.count+=val.count
});  
return res;
}

finalizeFun=function (key,reduceResult){
 reduceResult.amount=(reduceResult.amount).toFixed(2);
 reduceResult.avgPrice=(reduceResult.amount/reduceResult.count).toFixed(2); return reduceResult;}

 db.test.mapReduce(map,reduce,{out:"mr4",finalize:finalizeFun})
> db.mr4.find()
{ "_id" : "Joe", "value" : { "amount" : "2053.89", "count" : 395, "avgPrice" : "5.20" } }
{ "_id" : "Josh", "value" : { "amount" : "1409.26", "count" : 292, "avgPrice" : "4.83" } }
{ "_id" : "Ken", "value" : { "amount" : "1547.77", "count" : 313, "avgPrice" : "4.94" } }

5.統計単価が6より大きいSKUで、ユーザーごとの購入数.(データサブセットを選別してMRを行う)
  これは比較的に簡単です.MRを呼び出した時にスクリーニングクエリを加えるだけでいいです.他は変わりません.

db.test.mapReduce(map,reduce,{query:{price:{"$gt":6}},out:"mr5"})

締め括りをつける
   MongoDBのMRツールは非常に強力で、文中の例はあくまでも基本例であり、Shardingを結合した後、マルチサーバーが並列にデータセット処理をしてこそ、その能力が本当に現れます.