MongoDB获取数组的每个第N个元素答案

【问题标题】：MongoDB Get every Nth element of arrayMongoDB获取数组的每个第N个元素
【发布时间】：2020-07-21 03:24:21
【问题描述】：

我一直在从事一个小型项目，该项目从传感器获取 MQTT 数据并将其存储在 MongoDB 数据库中。我正在使用 nodeJS 和猫鼬。这些是我的架构。

export const SensorSchema = new mongoose.Schema({
    name: { type: String, required: true, unique: true },
    location: { type: String, required: true },
    type: { type: String, required: true },
    unit: { type: String, required: true },
    measurements: { type: [MeasurementSchema] }
},
{ 
    toObject: { virtuals: true },
    toJSON: { virtuals: true }
});

export const MeasurementSchema = new mongoose.Schema({
    value: {type: Number, required: true},
    time: {type: Date, required: true}
});

首先，我编写了一个函数，用于检索在两个时间戳之间进行的所有测量。

const values = Sensor.aggregate([
                { $match: Sensor.getValuesFromPath(sensorPath) },
                { $unwind: "$measurements"},
                { $match: { "measurements.time": { $gte: startTime,  $lte: endTime} }},
                { $replaceRoot: { newRoot: "$measurements" } },
                { $project: { _id: 0}},
                { $sort: {time: 1}}
            ]).exec();

为了在 UI 中绘制图表，我需要以某种方式排序然后限制发送到客户端的数据。我想在某个间隔内发送每个 Nth Value，以确保数据有点类似于数据的过程。我更喜欢不从数据库中获取所有数据的解决方案。

我将如何在数据库上执行此操作？我可以在排序后以某种方式访问元素的位置索引吗？ $arrayElemAt 或 $elemMatch 是解决方案吗？

【问题讨论】：

标签： node.js mongodb typescript mongoose aggregation-framework

【解决方案1】：

在运行$unwind 之前，您可以使用$filter 应用开始/结束日期过滤。这将允许您将measurements 处理为一个数组。在下一步中，您可以使用 $range 定义索引列表并使用 $arrayElemAt 从这些索引中检索元素来获取每个 N-th 元素：

const values = Sensor.aggregate([
                { $match: Sensor.getValuesFromPath(sensorPath) },                
                { $addFields: { 
                    measurements: { 
                        $filter: { 
                            input: "$measurements", 
                            cond: { $and: [ 
                                    { $gte: [ "$$this.time", startTime ] }, 
                                    { $lte: [ "$$this.time", endTime ] } 
                                ] 
                            } 
                        } 
                    } 
                } },
                { $addFields: { 
                    measurements: { 
                        $map: { 
                            input: input: { $range: [ 0, { $size: "$measurements" }, N ] }, 
                            as: "index",
                            in: { $arrayElemAt: [ "$measurements", "$$index" ] }
                        } 
                    } 
                } },
                { $unwind: "$measurements" },
                { $replaceRoot: { newRoot: "$measurements" } },
                { $project: { _id: 0}},
                { $sort: {time: 1}}
            ]).exec();

【讨论】：

$unwind 是一个非常昂贵的操作，或者您为什么更喜欢它在$unwind 之前完成所有过滤？否则它会不起作用吗？
@LucasEngleder no，$unwind 很好，您可以将数组元素作为单独的文档返回，但要获取 n-th 元素，您需要知道整个数据集的上下文，因此它必须是一个数组那时

【解决方案2】：

以下聚合 (i) 检索在两个时间戳之间进行的所有测量，(ii) 按每个传感器的时间戳排序，以及 (iii) 获取每第 N 个值（由变量 EVERY_N 指定）。

示例文档（带有一些任意数据用于测试）：

{
  name: "s-1",
  location: "123",
  type: "456",
  measurements: [ { time: 2, value: 12 }, { time: 3, value: 13 }, 
                  { time: 4, value: 15 }, { time: 5, value: 22 }, 
                  { time: 6, value: 34 }, { time: 7, value: 9 }, 
                  { time: 8, value: 5 }, { time: 9, value: 1 }, 
  ]
},
{
  name: "s-2",
  location: "789",
  type: "900",
  measurements: [ { time: 1, value: 31 }, { time: 3, value: 32 }, 
                  { time: 4, value: 35 }, { time: 6, value: 39 },
                  { time: 7, value: 6}, { time: 8, value: 70 }, 
                  { time: 9, value: 74 }, { time: 10, value: 82 }
  ]
}

聚合：

var startTime = 3, endTime = 10
var EVERY_N = 2   // value can be 3, etc.

db.collection.aggregate( [
  { 
      $unwind: "$measurements" 
  },
  { 
      $match: { 
          "measurements.time": { $gte: startTime, $lte: endTime } 
      } 
  },
  { 
      $sort: { name: 1, "measurements.time": 1 } 
  },
  { 
      $group: { 
          _id: "$name", 
          measurements: { $push: "$measurements" }, 
          doc: { $first: "$$ROOT" } 
      } 
  },
  { 
      $addFields: { 
          "doc.measurements": "$measurements" 
      } 
  },
  { 
      $replaceRoot: { newRoot: "$doc" } 
  },
  { 
      $addFields: { 
           measurements: { 
               $reduce: {
                    input: { $range: [ 0, { $size: "$measurements" } ] }, 
                    initialValue: [ ],
                    in: { $cond: [ { $eq: [ { $mod: [ "$$this", EVERY_N ] }, 0 ] }, 
                                   { $concatArrays: [ "$$value", [ { $arrayElemAt: [ "$measurements", "$$this" ] } ] ] }, 
                                   "$$value" 
                             ]
                    }
               }
           }
      } 
  }
] )

【讨论】：