title: Incremental MapReduce in c#.net
body: <p>I have data of following format..</p>
  
  <p>Type    C_ID    Assitor CollectionDate       granulity  counter</p>
  
  <p>A   a   10  08-08-2012 00:00    15  0.9912378</p>
  
  <p>B   a   5   08-08-2012 00:00    15  0.1860929</p>
  
  <p>C   b   4   08-08-2012 00:00    15  0.5345317</p>
  
  <p>D   c   1   08-08-2012 00:15    15  0.8656529</p>
  
  <p>E   b   1   08-08-2012 00:15    15  0.3249502</p>
  
  <p>A   a   10  08-08-2012 00:15    15  0.3743117</p>
  
  <p>B   a   5   08-08-2012 00:30    15  0.2608622</p>
  
  <p>C   b   4   08-08-2012 00:30    15  0.0079308</p>
  
  <p>D   c   1   08-08-2012 00:30    15  0.7094781</p>
  
  <p>E   b   1   08-08-2012 00:45    15  0.6133461</p>
  
  <p>A   a   10  08-08-2012 00:45    15  0.3035875</p>
  
  <p>B   a   5   08-08-2012 00:45    15  0.6093015</p>
  
  <p>C   b   4   08-08-2012 01:00    15  0.4104008</p>
  
  <p>D   c   1   08-08-2012 01:00    15  0.1687753</p>
  
  <p>E   b   1   08-08-2012 01:00    15  0.6627076</p>
  
  <p>A   a   10  08-08-2012 01:15    15  0.1901386
  .....</p>
  
  <p>and so on..</p>
  
  <p>I want to do incremental mapreduce on this table on hourly basis.. on CollectionDate is the field which show when the records are coming.. But all this code I want in c#.net</p>
  
  <p>I have done mapReduce but problem is I am getting 3 records every 15 minutes and 12Records every hour and after 1 hour these 15 records will be reduced.. and after next 1 hour again the rest of records will be reduced on same basis..</p>
  
  <p>Can I get help on this in c#.net only.. I am in deep trouble since last 20 Days..</p>
  
  <p>Its excess csv file.. from where i am getting the record.. n inserting into mongodb using c#.. in mongodb its look like sumthing this: 
  { "_id": a324b2f89d2e98fa21f, "Type" : A, "C_ID" : a, "assitor" : 10, "CollectionDate" : 08-08-2012 00:00, "Granulity" : 15, "Counter" : 0.1901386 } 
  { "_id": a324b2f89d2e98a216f, "Type" : B, "C_ID" : a, "assitor" : 10, "CollectionDate" : 08-08-2012 00:00, "Granulity" : 15, "Counter" : 0.1233542 } 
  { "_id": a324b2f89d2e98a3f2c, "Type" : A, "C_ID" : b, "assitor" : 12, "CollectionDate" : 08-08-2012 00:15, "Granulity" : 12, "Counter" : 0.8134552 }
  { "_id": a324b2f89d2e98b4e2d, "Type" : B, "C_ID" : b, "assitor" : 12, "CollectionDate" : 08-08-2012 00:15, "Granulity" : 12, "Counter" : 0.3218547 }</p>
  
  <p>OutputFile: 
  { "_id" : a8f3e231d456a675b23c, "CollectionDate": 08-08-2012 00:00 "AvgCounter" :  } 
  { "_id" : a8f3e232456a675a42cd, "CollectionDate": 08-08-2012 01:00 "AvgCounter" :  } 
  { "_id" : a8f3e231d46a67a0b4d2, "CollectionDate": 08-08-2012 02:00 "AvgCounter" :  } </p>
  
  <p>means hourly aggregated..</p>
  
  <p>Till What I have done...</p>
  
  <pre><code>private static void MapReduce(MongoDatabase db, String collName, BsonValue bsonValue, DateTime oldDateTime, DateTime newDateTime)
      {
          var collection = db.GetCollection&lt;BsonDocument&gt;(collName);
          Console.WriteLine(TotalReduction++); 
          String map = @"function() { 
                                  var sample = this;
                                  emit(sample.CollectionDate, {CID: sample.C_ID, count:1, CollectionTime: sample.CollectionDate});
                               }";
          String reduce = @"function(key, values) {
                                  var result = {CID: '', count:0};
                                  values.forEach(function(value){
                                      result.CID += value.CID;
                                      result.count += value.count;
                                      result.CollectionTime = value.CollectionTime;
                                  });
                                  return result;
                                  }";
          var options = new MapReduceOptionsBuilder();
          IMongoQuery[] queries = { Query.EQ("CollectionTime", bsonValue) };
          options.SetOutput(MapReduceOutput.Inline);
          IMongoQuery query = Query.And(queries);
          var results = collection.MapReduce(queries[0], map, reduce);
          collection = db.GetCollection&lt;BsonDocument&gt;("MSS_REDUCE");
          IEnumerable&lt;BsonDocument&gt; bdoc = results.GetResultsAs&lt;BsonDocument&gt;();
          collection.InsertBatch&lt;BsonDocument&gt;(bdoc);
      }
  </code></pre>
  
  <p>Thanks Ravi Sharma</p>