聚合展开文档键作为新文档

Question

我在更改使用 Mongo DB 构建的时间序列数据库的架构时遇到了一些问题。目前，我有如下所示的记录：

{
    "_id" : 20,
    "name" : "Bob,
    "location" : "London",
    "01/01/1993" : {
         "height" : "110cm",
         "weight" : "60kg",
    },
   "02/01/1993" : {
         "height" : "112cm",
         "weight" : "61kg",
    }

}

我希望使用聚合框架为每个“人”创建几条记录，为原始记录中的每个“时间-价值”子文档创建一条：

 {
    "_id" : 20,
    "name" : "Bob,
    "date" : "01/01/1993"
    "location" : "London",
    "height" : "110cm",
    "weight" : "60kg",
 },

 {
    "_id" : 20,
    "name" : "Bob,
    "date" : "02/01/1993"
    "location" : "London",
    "height" : "112cm",
    "weight" : "61kg",
 }

在向每条记录添加大量时间序列值时，新方案应该更有效，而且我不应该遇到最大文档大小错误！

任何有关如何使用 Mongo DB 聚合管道执行此操作的帮助将不胜感激！

Answer 1

虽然 Aggregation Framework 的现代版本中有一些功能可以让您做这种事情，但它是否真的是最好的解决方案可能会有所不同。

本质上，您可以创建一个条目数组，其中包含“不包括”其他顶级键的文档键，这些键随后将包含在文档中。然后可以使用$unwind 处理该数组，并将整个结果重新整形为新文档：

db.getCollection('input').aggregate([
  { "$project": {
    "name": 1,
    "location": 1,
    "data": {
      "$filter": {
        "input": { "$objectToArray": "$$ROOT" },
        "as": "d",
        "cond": {
          "$not": { "$in": [ "$$d.k", ["_id","name","location"] ] }    
        }
      }  
    }  
  }},
  { "$unwind": "$data" },
  { "$replaceRoot": {
    "newRoot": {  
      "$arrayToObject": {
        "$concatArrays": [  
          [{ "k": "id", "v": "$_id" },
           { "k": "name", "v": "$name" },
           { "k": "location", "v": "$location" },
           { "k": "date", "v": "$data.k" }],
          { "$objectToArray": "$data.v" }
        ]
      }
    }
  }},
  { "$out": "output" }
])

或者在生成的数组元素中对初始 $project 进行所有重塑：

db.getCollection('input').aggregate([
  { "$project": {
    "_id": 0,
    "data": {
      "$map": {
        "input": {
          "$filter": {
            "input": { "$objectToArray": "$$ROOT" },
            "as": "d",
            "cond": {
              "$not": { "$in": [ "$$d.k", ["_id", "name", "location"] ] }    
            }
          }
        },
        "as": "d",
        "in": {
          "$arrayToObject": {
            "$concatArrays": [
              { "$filter": {
                "input": { "$objectToArray": "$$ROOT" },
                "as": "r",
                "cond": { "$in": [ "$$r.k", ["_id", "name", "location"] ] }
              }},
              [{ "k": "date", "v": "$$d.k" }],
              { "$objectToArray": "$$d.v" }
            ]   
          }
        }
      }
    }  
  }},
  { "$unwind": "$data" },
  { "$replaceRoot": { "newRoot": "$data" } },
  { "$out": "output" }
])

因此，您使用$objectToArray 和$filter 来根据实际包含每个日期的数据点的键创建一个数组。

在$unwind 之后，我们基本上将$arrayToObject 应用于“数组格式”的一组命名键，以便为$replaceRoot 构造newRoot，然后写入新集合，作为一个新文档每个数据键使用$out。

不过，这可能只会让您有所收获，因为您确实应该将 "date"data 更改为 BSON 日期。它占用的存储空间更少，也更容易查询。

var updates = [];
db.getCollection('output').find().forEach( d => {
  updates.push({
    "updateOne": {
      "filter": { "_id": d._id },
      "update": {
        "$set": {
          "date": new Date(
            Date.UTC.apply(null,
              d.date.split('/')
                .reverse().map((e,i) => (i == 1) ? parseInt(e)-1: parseInt(e) )
            )
          )
        }
      }
    }
  });
  if ( updates.length >= 500 ) {
    db.getCollection('output').bulkWrite(updates);
    updates = [];
  }
})

if ( updates.length != 0 ) {
  db.getCollection('output').bulkWrite(updates);
  updates = [];
}

当然，如果您的 MongoDB 服务器缺少这些聚合功能，那么您最好首先通过迭代循环将输出写入新集合：

var output = [];

db.getCollection('input').find().forEach( d => {
  output = [
    ...output,
    ...Object.keys(d)
      .filter(k => ['_id','name','location'].indexOf(k) === -1)
      .map(k => Object.assign(
        { 
          id: d._id,
          name: d.name,
          location: d.location,
          date: new Date(
            Date.UTC.apply(null,
              k.split('/')
                .reverse().map((e,i) => (i == 1) ? parseInt(e)-1: parseInt(e) )
            )
          )
        },
        d[k]
      ))
  ];

  if ( output.length >= 500 ) {
    db.getCollection('output').insertMany(output);
    output = [];    
  }
})

if ( output.length != 0 ) {
  db.getCollection('output').insertMany(output); 
  output = [];
}

在任何一种情况下，我们都希望将Date.UTC 应用于现有“字符串”基于日期的反转字符串元素，并获得一个可以转换为 BSON 日期的值。

聚合框架本身不允许类型转换，因此该部分（并且它是必要部分）的唯一解决方案是实际循环和更新，但是使用表单至少可以提高循环和更新的效率。

任何一种情况都会给你相同的最终输出：

/* 1 */
{
    "_id" : ObjectId("599275b1e38f41729f1d64fe"),
    "id" : 20.0,
    "name" : "Bob",
    "location" : "London",
    "date" : ISODate("1993-01-01T00:00:00.000Z"),
    "height" : "110cm",
    "weight" : "60kg"
}

/* 2 */
{
    "_id" : ObjectId("599275b1e38f41729f1d64ff"),
    "id" : 20.0,
    "name" : "Bob",
    "location" : "London",
    "date" : ISODate("1993-01-02T00:00:00.000Z"),
    "height" : "112cm",
    "weight" : "61kg"
}