如何使用聚合查询sales集合中连续价格上涨的商品销售记录？

需求说明 #

我们需要从sales集合中筛选出存在连续价格上涨的商品的对应上涨记录（即后一条记录价格高于同商品的前一条记录）。示例中符合条件的是Banana价格为6的那条记录，而Melon因只有单条记录、Pineapple后续价格下跌，均不符合要求。

集合原始数据 #

[
  {
    product: "Banana",
    timestamp: 1672992000,
    price: 5,
  },
  {
    product: "Banana",
    timestamp: 1672992001,
    price: 6,
  },
  {
    product: "Pineapple",
    timestamp: 1672992000,
    price: 9,
  },
  {
    product: "Pineapple",
    timestamp: 1672992001,
    price: 8,
  },
  {
    product: "Melon",
    timestamp: 1672992005,
    price: 15,
  }
]

聚合实现方案 #

完全可以通过MongoDB的聚合管道实现，核心思路是：按商品分组，将同商品的记录按时间排序，对比相邻记录的价格，筛选出价格上涨的目标记录。

具体聚合管道代码如下：

db.sales.aggregate([
  // 1. 按商品分组，收集同商品的所有记录并保存原始文档
  {
    $group: {
      _id: "$product",
      records: {
        $push: {
          timestamp: "$timestamp",
          price: "$price",
          original: "$$ROOT"
        }
      }
    }
  },
  // 2. 对同商品的记录按时间升序排序，保证连续顺序正确
  {
    $set: {
      records: {
        $sortArray: {
          input: "$records",
          sortBy: { timestamp: 1 }
        }
      }
    }
  },
  // 3. 生成当前记录与前一条记录的配对数组
  {
    $set: {
      recordPairs: {
        $map: {
          input: { $range: [1, { $size: "$records" }] },
          as: "index",
          in: {
            current: { $arrayElemAt: ["$records", "$$index"] },
            previous: { $arrayElemAt: ["$records", { $subtract: ["$$index", 1] }] }
          }
        }
      }
    }
  },
  // 4. 筛选出当前价格高于前一条的记录配对
  {
    $set: {
      increasedPairs: {
        $filter: {
          input: "$recordPairs",
          cond: { $gt: ["$$this.current.price", "$$this.previous.price"] }
        }
      }
    }
  },
  // 5. 只保留存在价格上涨记录的商品数据
  {
    $match: {
      $expr: { $gt: [{ $size: "$increasedPairs" }, 0] }
    }
  },
  // 6. 展开并提取符合条件的原始记录
  {
    $unwind: "$increasedPairs"
  },
  {
    $replaceRoot: { newRoot: "$increasedPairs.current.original" }
  }
])

代码阶段说明 #

• $group：按商品维度聚合，收集同商品的所有销售记录，同时保留原始文档用于最终输出。
• $sortArray：确保同商品的记录按时间先后排序，保证相邻记录是连续的销售节点。
• $map+$range：从第2条记录开始，生成当前记录与前序记录的配对，为价格对比做准备。
• $filter：筛选出当前价格高于前序价格的记录配对。
• $match：过滤掉没有价格上涨记录的商品分组。
• $unwind+$replaceRoot：将符合条件的原始记录展开并作为根文档输出，得到最终目标结果。

执行结果 #

运行上述聚合后，将得到如下结果：

{
  "_id": ObjectId("..."), // 原始文档的ID
  "product": "Banana",
  "timestamp": 1672992001,
  "price": 6
}

内容的提问来源于stack exchange，提问作者Nerver Corameiro