一、Aggregate簡介                                                                           

      db.collection.aggregate()是基於資料處理的聚合管道，每個文件通過一個由多個階段（stage）組成的管道，可以對每個階段的管道進行分組、過濾等功能，然後經過一系列的處理，輸出相應的結果。

      1、db.collection.aggregate()可以多個管道，能方便的進行資料的處理。

      2、db.collection.aggregate()使用了MongoDB

      3、 每個階段管道限制為100MB的記憶體。如果一個節點管道超過這個極限,MongoDB將產生一個錯誤。為了能夠在處理大型資料集,可以設定allowDiskUse為true來在聚合管道節點把資料寫入臨時檔案。這樣就可以解決100MB的記憶體的限制。

     4、db.collection.aggregate()可以作用在分片集合，但結果不能輸在分片集合，MapReduce可以 作用在分片集合，結果也可以輸在分片集合。

     5、db.collection.aggregate()

     6、db.collection.aggregate()輸出的結果只能儲存在一個文件中，BSON Document大小限制為16M。可以通過返回指標解決，版本2.6中後面：DB.collect.aggregate()方法返回一個指標，可以返回任何結果集的大小。

二、Aggregate pipeline 引數

          今天主要是pipeline 運用到Spring上：

   【pipeline 引數】

     pipeline 型別是Array  語法：db.collection.aggregate( [ { <stage> }, ... ] )

         $project：可以對輸入文件進行新增新欄位或刪除現有的欄位，可以自定哪些欄位顯示與不顯示。

         $match ：根據條件用於過濾資料，只輸出符合條件的文件，如果放在pipeline前面，根據條件過濾資料，傳輸到下一個階段管道，可以提高後續的資料處理效率。還可以放在out之前，對結果進行再一次過濾。

         $limit ：用來限制MongoDB聚合管道返回的文件數

         $skip :在聚合管道中跳過指定數量的文件，並返回餘下的文件。

         $unwind :將文件中的某一個數組型別欄位拆分成多條，每條包含陣列中的一個值。    

         $out :必須為pipeline最後一個階段管道，因為是將最後計算結果寫入到指定的collection中。

         $group : 將集合中的文件分組，可用於統計結果，$group首先將資料根據key進行分組。

       MongoDB上對 pipeline 操作可以檢視

三、舉例子並在MongoDB操作pipeline                                                                                                                      

db. orders.insert([  
{  
        "onumber" : "001",   
        "date" : "2015-07-02",   
        "cname" : "zcy1",   
         "items" :[ {  
                   "ino" : "001",  
                  "quantity" :2,   
                  "price" : 4.0  
                 },{  
                   "ino" : "002",  
                  "quantity" : 4,   
                  "price" : 6.0  
                }  
                ]  
},{  
         "onumber" : "002",   
        "date" : "2015-07-02",   
        "cname" : "zcy2",   
         "items" :[ {  
                  "ino" : "003",  
                  "quantity" :1,   
                  "price" : 4.0  
                   },{  
                  "ino" : "002",  
                  "quantity" :6,   
                  "price" : 6.0  
                 }  
               ]  
},{  
         "onumber" : "003",   
        "date" : "2015-07-02",   
        "cname" : "zcy2",   
         "items" :[ {  
                  "ino" : "004",  
                  "quantity" :3,   
                  "price" : 4.0  
                   },{  
                  "ino" : "005",  
                  "quantity" :1,   
                  "price" : 6.0  
                 }  
               ]  
},{  
         "onumber" : "004",   
        "date" : "2015-07-02",   
        "cname" : "zcy2",   
         "items" :[ {  
                  "ino" : "001",  
                  "quantity" :3,   
                  "price" : 4.0  
                   },{  
                  "ino" : "003",  
                  "quantity" :1,   
                  "price" : 6.0  
                 }  
               ]  
}
])

> db.orders.aggregate([
... {$match:{"onumber":{$in:["001","002", "003"]}}},
... {$unwind:"$items"},
... {$group:{_id:"$items.ino",total:{$sum:"$items.quantity"}}},
... {$match:{total:{$gt:1}}}
... ]);
{ "_id" : "004", "total" : 3 }
{ "_id" : "002", "total" : 10 }
{ "_id" : "001", "total" : 2 }

四、AggregationOutput                                                                                                                

             Spring Data  MongoDB 專案提供與MongoDB文件資料庫的整合。如果對環境的搭建不清楚可以先看

        我們先介紹Mongo本身提供的com.mongodb.AggregationOutput進行分組查詢，下一篇會介紹Spring Data MongoDB模板封裝的Aggregation方法，我們直接看程式碼

	@Override
	public void getAggregation() {
		Set<String> onumberSet=new HashSet<String>();
		onumberSet.add("001");
		onumberSet.add("002");
		onumberSet.add("003");
		//過濾條件
		DBObject queryObject=new BasicDBObject("onumber", new BasicDBObject("$in",onumberSet));
		DBObject queryMatch=new BasicDBObject("$match",queryObject);
		//展開陣列
		DBObject queryUnwind=new BasicDBObject("$unwind","$items");
		//分組統計
		DBObject groupObject=new BasicDBObject("_id",new BasicDBObject("ino","$items.ino"));
		groupObject.put("total", new BasicDBObject("$sum","$items.quantity"));
		DBObject  queryGroup=new BasicDBObject("$group",groupObject);
		//過濾條件
		DBObject finalizeMatch=new BasicDBObject("$match",new BasicDBObject("total",new BasicDBObject("$gt",1)));

		AggregationOutput  output=mongoTemplate.getCollection("orders").aggregate(queryMatch,queryUnwind,queryGroup,finalizeMatch);
		for (Iterator<DBObject> iterator = output.results().iterator(); iterator.hasNext();) {
			DBObject obj =iterator.next();
			System.out.println(obj.toString());
		}
          }

public class TestOrders {
	 private static OrdersDao ordersDaoImpl;  
	    private static  ClassPathXmlApplicationContext  app;  
	    @BeforeClass    
	    public static void initSpring() {   
	        try {         
	         app = new ClassPathXmlApplicationContext("classpath:applicationContext-mongo.xml");    
	         ordersDaoImpl = (OrdersDao) app.getBean("ordersDaoImpl");   
	        } catch (Exception e) {  
	            e.printStackTrace();  
	        }  
	           }   
	      
	        @Test   
	        public void testAggregation()  
	        {  
	        	ordersDaoImpl.getAggregation();
	            
	        }  
}