mongodb的聚合子查詢：MongoDB的PHP的適配器

Question

我有以下集合：

**S_SERVER** – **S_PORT** – **D_PORT** – **D_SERVER** – **MBYTES** 
L0T410R84LDYL – 2481 – 139 – MRMCRUNCHAPP – 10 
MRMCASTLE – 1904 – 445 – MRMCRUNCHAPP – 25 
MRMXPSCRUNCH01 – 54769 – 445 – MRMCRUNCHAPP - 2 
MRMCASTLE – 2254 – 139 – MRMCRUNCHAPP - 4 
MRMCASTLE – 2253 – 445 – MRMCRUNCHAPP -35 
MRMCASTLE – 987 – 445 – MRMCRUNCHAPP – 100 
MRMCASTLE – 2447 – 445 – MRMCRUNCHAPP – 12 
L0T410R84LDYL – 2481 – 139 – MRMCRUNCHAPP - 90 
MRMCRUNCHAPP – 61191 – 1640 – OEMGCPDB – 10 

首先，我需要前30 S_SERVER按照從每個S_SERVER轉移總字節。這是我能夠獲得與下面的查詢：

$sourcePipeline = array(
     array(
      '$group' => array(
       '_id' => array('sourceServer' => '$S_SERVER'), 
       'MBYTES' => array('$sum' => '$MBYTES') 
      ), 
     ), 
     array(
      '$sort' => array("MBYTES" => -1), 
     ), 
     array(
      '$limit' => 30 
     ) 
    ); 
$sourceServers = $collection->aggregate($sourcePipeline); 

我還需要30強D端口按每個D端口個人S_SERVER上傳輸的總字節。我通過從上面的服務器結果運行for循環並逐個爲每個S_SERVER逐個獲取它們來完成此操作。

$targetPortPipeline = array(
       array(
        '$project' => array('S_SERVER' => '$S_SERVER', 'D_PORT' => '$D_PORT', 'MBYTES' => '$MBYTES') 
       ), 
       array(
        '$match' => array('S_SERVER' => S_SERVER(find from above query, passed one by one in for loop)), 
       ), 
       array(
        '$group' => array(
         '_id' => array('D_PORT' => '$D_PORT'), 
         'MBYTES' => array('$sum' => '$MBYTES') 
        ), 
       ), 
       array(
        '$sort' => array("MBYTES" => -1), 
       ), 
       array(
        '$limit' => $limit 
       ) 
      ); 
$targetPorts = $collection->aggregate($targetPortPipeline); 

但是這個過程花費了太多時間。我需要一種有效的方式來在相同的查詢中獲得所需的結果。我知道我使用Mongodb的PHP適配器來實現這一點。你也可以讓我知道javascript格式的聚合函數。我將它轉換成php。

Answer 1

你這裏的問題基本上似乎是你發出30個更多的查詢爲您的最初30個結果。對此沒有簡單的解決方案，目前單個查詢似乎正確，但您可以考慮幾件事情。

作爲補充說明，您並不孤單，因爲這是我之前見過的一個問題，我們可以將其稱爲「排名前N的結果問題」。實質上，您真正想要的是將兩個結果集組合在一起，以便每個分組邊界（源服務器）本身的結果最大值僅爲N，而在頂層時，您也將這些結果再次限制爲頂部的N結果值。

您的第一個聚合查詢您所需的前30個「源服務器」的結果，這就好了。但是，而不是從這個循環其他查詢，你可以嘗試只是「源服務器」從這個結果值創建一個數組，並使用$in操盤手傳遞，爲您的第二個查詢：

db.collection.aggregate([ 
    // Match should be first 
    { "$match": { "S_SERVER": { "$in": sourceServers } } }, 

    // Group both keys 
    { "$group": { 
     "_id": { 
      "S_SERVER": "$S_SERVER", 
      "D_SERVER": "$D_SERVER" 
     }, 
     "value": { "$sum": "$MBYTES" } 
    }}, 

    // Sort in order of key and largest "MBYTES" 
    { "$sort": { "S_SERVER": 1, "value": -1 } } 
]) 

注意到你不能在這裏「限制」，因爲這包含了最初匹配的每個「源服務器」。您也可以不在分組邊界上「加以限制」，這本質上是聚合框架中所缺少的，以使其成爲兩個查詢結果。

由於這包含每個「dest服務器」結果並可能超出「前30名」，因此您將在代碼中處理結果並在每個分組（源服務器）級別檢索「前30名」後跳過返回的結果。取決於你有多少結果，這可能是也可能不是最實際的解決方案。

繼續前進，如果這不太實際，那麼你會非常難以將該輸出作爲臨時步驟輸入到另一個集合中。如果你有一個MongoDB 2.6或更高版本，這可以像在聲明結尾處添加一個流水線階段那樣簡單。對於早期版本，你可以做等同語句中使用的MapReduce：

db.collection.mapReduce(
    function() { 
     emit(
      { 
       "S_SERVER": this["S_SERVER"], 
       "D_SERVER": this["D_SERVER"] 
      }, 
      this.MBYTES 
     ); 
    }, 
    function(key,values) { 
     return Array.sum(values); 
    }, 
    { 
     "query": { "S_SERVER": { "$in": sourceServers } }, 
     "out": { "replace": "output" } 
    } 
) 

這本質上是相同的過程，因爲前一個聚合的語句，同時還指出，mapReduce不輸出排序。這就是由附加mapReduce操作覆蓋在所獲得的收集：

db.output.mapReduce(
    function() { 

     if (cServer != this._id["S_SERVER"]) { 
      cServer = this._id["S_SERVER"]; 
      counter = 0; 
     } 

     if (counter < 30) 
      emit(this._id, this.value); 

     counter++; 
    }, 
    function(){}, // reducer is not actually called 
    { 
     "sort": { "_id.S_SERVER": 1, "value": -1 }, 
     "scope": { "cServer": "", "counter": 0 } 
    } 
) 

這裏的實現是前面提到的「光標跳躍」的「服務器端」版本。因此，您仍然在處理每個結果，但通過網絡返回的結果僅限於每個「源服務器」下的前30個結果。

如前所述，仍然是相當可怕的這個必須掃描「編程」通過結果丟棄你不想要的孩子，和捲上同樣取決於，你可能會簡單地發放AA .find()每個更好在這些結果中的「源服務器」價值的同時，排序和限制結果

sourceServers.forEach(function(source) { 
    var cur = db.output.find({ "_id.S_SERVER": source }) 
     .sort({ "value": -1 }).limit(30); 
    // do something with those results 
); 

這仍然是額外的30個查詢，但至少你不是「聚集」每一次爲工作已經完成。

作爲最後一點說明，實際上需要詳細瞭解詳情，可以使用顯示的初始聚合查詢的修改形式來處理此問題。這更像是一個腳註，如果你已經閱讀了這麼多，而其他方法看起來並不合理，那麼這可能是最糟糕的，因爲這可能會造成內存限制。

引進最好的方法是用「理想」的情況下爲「前N個結果」聚集，這當然實際上並不存在，但理想的管道末端會是這個樣子：

{ "$group": { 
     "_id": "$S_SERVER", 
     "results": { 
      "$push": { 
       "D_SERVER": "$_id.D_SERVER", 
       "MBYTES": "$value" 
      }, 
      "$limit": 30 
     } 
    }} 

因此，這裏的「不存在」因素是「限制」每個「源服務器」值「推」到結果數組中的結果數量的能力。如果實現了這個或類似的功能，那麼這個世界肯定會變得更好，因爲它可以很容易地解決問題。

由於它不存在，你只能採取其他方法來獲得相同的結果，並最終得到像this example這樣的列表，除了在這種情況下實際上更復雜。

考慮到代碼，你會做線沿線的東西：

1. 集團所有結果返回到每個服務器的陣列
2. 集團所有這一切回到一個單一的文件
3. 解開第一個服務器結果
4. 獲取第一個條目並再次返回組。
5. 放鬆結果又一次
6. 項目並匹配找到的條目
7. 放棄匹配結果
8. 沖洗，然後重複步驟4 - 7 30倍
9. 商店後面的30個結果
第一服務器文檔
10. 沖洗和重複爲3 - 9對於每個服務器，所以30倍

你永遠不會直接編寫代碼，而必須編寫代碼生成管道階段。這很可能會炸燬16MB的限制，可能不在管道文檔本身上，但很可能在實際結果集上這樣做，因爲您正在將所有內容都推送到數組中。

您可能還會注意到，如果您的實際結果不包含每臺服務器至少30個最高值，則完全炸燬是多麼容易。

整個局面歸結爲哪種方法適合你的數據和性能方面的考慮最好的權衡：

1. 現場與您最初的結果30個聚集查詢。
2. 減少爲2個查詢並丟棄客戶端代碼中不需要的結果。
3. 輸出到臨時集合並使用服務器光標跳過以放棄結果。
4. 從預先彙總的集合中發出30個查詢。
5. 實際上遇到了執行產生結果的管線的麻煩。

在任何情況下，由於總的來說很複雜，我會定期生成結果集並將其存儲在自己的集合中，而不是試圖實時執行此操作。

這些數據不會是「最新」的結果，而只是您更新頻率的最新結果。但實際上，檢索這些結果以供顯示變成一個簡單的查詢，最多返回900個相當緊湊的結果，而不會爲每個請求進行彙總。