$project MongoDB 聚合中的不同项目

Question

我在 mongodb 中有一个非常大的项目集合，其架构无法更改。简化版如下所示：

{event: { address: {ip: "1.1.1.1", port: 80}}}
{event: { address: {ip: "1.1.1.2", port: 80}}}
{event: { address: [{ip: "1.1.1.1", port: 80}, {ip: "1.1.1.1", port: 443}]}}
{event: { address: [{ip: "1.1.1.1", port: 8080}, {ip: "1.1.1.2", port: 443}]}}

每个事件可能有一个或多个地址。每个地址都有“ip”和“port”。因此，一个“ip”可能会在具有多个地址的事件中重复。

我要做的就是统计每个 ip 地址的事件数并找到排名靠前的 ip 地址。对于上面的例子，首选的结果是：

[ { "ip" : "1.1.1.1", "count" : 3 },
  { "ip" : "1.1.1.2", "count" : 2 } ]

想到的一个查询是这样的：

db.collection.aggregate({$project: {ip: "$event.address.ip"}}, {$group: {_id: "$ip", count: {$sum: 1}}}, {$sort: {count: -1}}, {$limit: 5})

但结果是：

{
 "result" : [
  { "_id" : ["1.1.1.1", "1.1.1.2"], "count" : 1 },
  { "_id" : ["1.1.1.1", "1.1.1.1"], "count" : 1 },
  { "_id" : "1.1.1.2", "count" : 1 },
  { "_id" : "1.1.1.1", "count" : 1 } ],
 "ok" : 1
}

我不能使用 $unwind，因为每个 IP 地址对于每个事件都应该只计算一次，但有些事件会重复相同的 IP。此外， $unwind 通常不会起作用，因为“地址”并不总是一个数组。有些事件只有一个不是数组的地址，$unwind 会为它们抛出异常。

我尝试了不同的聚合运算符，例如 $group 中的 $addToSet，但都无济于事。

集合非常庞大，我无法先提取应用程序中的所有 ip 地址，然后为每个事件统计事件。

可以用 map/reduce 来完成吗？你有什么建议？

Answer 1

虽然这可以通过 MapReduce 完成，但聚合框架会更快。您需要在计划中添加两个步骤 - 1）您需要“规范化”格式，以便地址始终是一个数组，2）然后您需要 $unwind 该数组，按 _id,ip 分组以消除重复项和然后按 ip 分组以获得您需要的计数。

规范化数组和非数组很棘手，但可以通过$unwind 之前和之后的两个投影来完成。

var p1 = { "$project" : {
        "array" : {
            "$cond" : [
                {
                    "$eq" : [
                        "$address.0",
                        [ ]
                    ]
                },
                "$address",
                [
                    null
                ]
            ]
        },
        "notarray" : {
            "$cond" : [
                {
                    "$ne" : [
                        "$address.0",
                        [ ]
                    ]
                },
                "$address",
                [
                    null
                ]
            ]
        },
        "isArray" : {
            "$eq" : [
                "$address.0.ip",
                [ ]
            ]
        }
    }
};
var u = { "$unwind" : "$array" };
var p2 = { "$project" : {
        "address" : {
            "$cond" : [
                "$isArray",
                "$array",
                "$notarray"
            ]
        }
    }
};

相比之下，$group 的两个阶段比较简单：

var g1 = { "$group" : { "_id" : { "_id" : "$_id", "ip" : "$address.ip" } } };
var g2 = { "$group" : { "_id" : "$_id.ip", "count" : { "$sum" : 1 } } };

这是我的示例数据：

> db.coll.find()
{ "_id" : ObjectId("52cd0badba17f3b7ed212575"), "address" : { "ip" : "1.1.1.1" } }
{ "_id" : ObjectId("52cd0bc4ba17f3b7ed212576"), "address" : [  {  "ip" : "1.1.1.1" },  {  "ip" : "1.1.1.1" } ] }
{ "_id" : ObjectId("52cd0bc9ba17f3b7ed212577"), "address" : [  {  "ip" : "1.1.1.1" },  {  "ip" : "1.1.1.2" } ] }

这里是聚合及其输出：

> db.coll.aggregate(p1, u, p2, g1, g2)
{ "_id" : "1.1.1.1", "count" : 3 }
{ "_id" : "1.1.1.2", "count" : 1 }

Answer 2

啊，你有两个问题，一个是架构设计得不好，二是它没有标准化，因此同一个字段至少没有相同的结构。你在岩石和坚硬的地方之间。

如果所有地址字段都是数组，那么这将很容易工作，但是您不能有条件地当前$unwind，不幸的是，如果您尝试$unwind 除了数组之外的任何内容，以便您可以展开地址，那么您将收到错误：

如果您为 $unwind 指定的目标字段不是数组，则 db.collection.aggregate() 会生成错误。

http://docs.mongodb.org/manual/reference/operator/aggregation/unwind/

所以是的，你的那种卡在这里。

这可以通过 MR 完成，但分组会很麻烦。就我个人而言，我要做的是运行增量 MR，它以标准化格式将此模式写出，以便可以在其上使用聚合框架。