合并重复项并删除最旧的

Question

我有一个集合，其中有一些重复的文档。例如：

第一个文档：

{
    "_id" : ObjectId("56f3d7cc1de31cb20c08ae6b"),
    "AddedDate" : ISODate("2016-05-01T00:00:00.000Z"),
    "Place": "THISPLACE",
    "PresentInDB" : [ 
        {
            "InDB" : ISODate("2016-05-01T00:00:00.000Z")
        }
     ],
    "Checked" : [],
    "Link": "http://www.mylink.com/first/84358"
}

第二个文档：

{
    "_id" : ObjectId("577740526c1e542904725238"),
    "AddedDate" : ISODate("2016-05-02T00:00:00.000Z"),
    "Place": "THISPLACE",
    "PresentInDB" : [ 
        {
            "InDB" : ISODate("2016-05-02T00:00:00.000Z")
        },
        {
            "InDB" : ISODate("2016-05-03T00:00:00.000Z")
        }
     ],
    "Checked" : [
        {
            "Done" : ISODate("2016-05-02T00:00:00.000Z")
        },
    ],
    "Link": "http://www.mylink.com/second/84358"
}

Link 字段在两个文档中包含相同的数字序列，84358。

所以我想实现这些步骤：

遍历集合中的每个文档。
匹配Link字段（即上面的84358）每个文档中的编号顺序，如果有几个个文档 Link 字段中具有该序列的集合。并且如果 Place 两个文档中的字段匹配：
合并 PresentInDB 和 Checked 字段 -> 通过添加最新文档中的数组值（按 AddedDate 中的日期合并 PresentInDB 和 Checked 字段字段）到最旧的文档。
删除最新的文档。

如何实现这样的查询？

Answer 1

您可以使用单个 aggregation 查询来完成此操作：

db.device.aggregate([{
    "$unwind": "$PresentInDB"
}, {
    "$match": {
        "Link": /84358/
    }
}, {
    "$sort": {
        "AddedDate": 1
    }
}, {
    "$group": {
        _id: 0,
        PresentInDB: {
            $addToSet: '$PresentInDB'
        },
        AddedDate: {
            $first: "$AddedDate"
        },
        id: {
            $first: "$_id"
        },
        Link: {
            $first: "$Link"
        }
    }
}, {
    $out: "documents"
}])

$unwind 你的数组可以处理它
$match你的id（这里包含84358）
$sort 按日期升序
$group 与 :
- a $addToSet 将所有 PresentInDB 合并到一个没有重复的数组中
- a $first 为每个要保留的字段。保留第一个意味着你只想要旧的，因为我们之前按升序日期排序
$out 会将结果保存到一个名为 documents 的新集合中

Answer 2

在 MongoDB 3.3.6 版本中引入了一个 $split 运算符，用于在聚合框架 (Jira) 中处理字符串。在此版本之前，您只能使用 map/reduce 解决方案解决此问题。

MongoDB3.3.6发布后：聚合框架解决方案

db.duplicatedCollection.aggregate(
  [
    {
      $project: {
        _id : 1,
        AddedDate : 1,
        Place : 1,
        PresentInDB : 1,
        Checked : 1,
        Link : 1,
        sequenceNumber: { $arrayElemAt: [ {$split: ["$Link", "/"]}, -1 ]},
      }
    }, 
    {
      $sort: { AddedDate: 1 }
    },
    {
      $group: {
        _id : {
          sequenceNumber : "$sequenceNumber",
          Place : "$Place"
        },
        id : { $first: "$_id"},
        AddedDate: { $first: "$AddedDate" },
        Place :  { $first: "$Place" },
        PresentInDB: {
          $push: '$PresentInDB'
        },
        Checked: {
          $push: '$Checked'
        },
        Link: { $first: "$Link"}
      }
    }, 
   {
      $unwind: "$PresentInDB"
    },
    {
      $unwind: {
        path : "$PresentInDB",
        preserveNullAndEmptyArrays: true
      }    
    },
    {
      $unwind: "$Checked"
    },
    {
      $unwind: {
        path : "$Checked",
        preserveNullAndEmptyArrays: true
      }
    },    
    {
      $group: {
        _id : "$id",
        AddedDate: { $first: "$AddedDate" },        
        Place :  { $first: "$Place" },
        PresentInDB : {
          $addToSet: '$PresentInDB'
        },
        Checked : {
          $addToSet: '$Checked'
        },        
        Link: { $first: "$Link"}
      }
    },  
    {
      $out: "duplicatedCollection"
    }
  ]
);

MongoDB3.3.6 发布之前：Map/Reduce解决方案

地图函数：

var mapFunction = function() {
    var linkArray = this.Link.split("/");
    var sequenceNumber = linkArray[linkArray.length - 1];

    var keyDoc = { 
       place : this.Place,
       sequenceNumber: sequenceNumber, 
    };

    emit(keyDoc, this);
};

归约函数：

var reduceFunction = function(key, values) {   
    var reducedDoc = {};
    reducedDoc._id = values[0]._id;
    reducedDoc.AddedDate = values[0].AddedDate;
    reducedDoc.Link = values[0].Link;
    reducedDoc.PresentInDB = [];
    reducedDoc.Checked = [];

    var presentInDbMillisArray = [];
    var checkedMillisArray = [];        

    values.forEach(function(doc) {
        if (reducedDoc.AddedDate < doc.AddedDate) {
            reducedDoc._id = doc._id;
            reducedDoc.AddedDate = doc.AddedDate;
            reducedDoc.Link = doc.Link;
        } 

        // PresentInDB field merge
        doc.PresentInDB.forEach(function(presentInDBElem) {
            var millis = presentInDBElem.InDB.getTime();
            if (!Array.contains(presentInDbMillisArray, millis)) { 
                reducedDoc.PresentInDB.push(presentInDBElem);
                presentInDbMillisArray.push(millis);
            }
        });

        // same here with Checked field
        doc.Checked.forEach(function(checkedElem) {
            var millis = checkedElem.Done.getTime();
            if (!Array.contains(checkedMillisArray, millis)) { 
                reducedDoc.Checked.push(checkedElem);
                checkedMillisArray.push(millis);
            }
        });
    });
    return reducedDoc;
};

Map/Reduce:

db.duplicatedCollection.mapReduce(
    mapFunction,
    reduceFunction,
    { 
        "out": "duplicatedCollection"
    }
);

从 map/reduce 返回的文档中展开值：

   db.duplicatedCollection.find(
    {
        value : {
            $exists: true 
        }
    }
    ).forEach(function(doc) {
        db.duplicatedCollection.insert(doc.value);
        db.duplicatedCollection.remove({_id : doc._id});
    });

合并重复项并删除最旧的

Merge duplicates and remove the oldest

merge

mapreduce

updates

mongodb