Amazon DocumentDB 聚合管道 #

聚合管道：
├── 数据处理管道
├── 多个阶段顺序执行
├── 每个阶段处理并传递结果
├── 支持复杂的数据转换
└── 类似Linux管道操作

db.collection.aggregate([
  { stage1 },
  { stage2 },
  { stage3 }
])

常用聚合阶段：
├── $match - 过滤
├── $group - 分组
├── $project - 投影
├── $sort - 排序
├── $limit - 限制数量
├── $skip - 跳过
├── $unwind - 展开数组
├── $lookup - 关联查询
├── $addFields - 添加字段
└── $facet - 多管道并行

// 过滤文档
db.orders.aggregate([
  {
    $match: {
      status: "completed",
      amount: { $gte: 100 }
    }
  }
])

// 尽早使用$match减少数据量
db.orders.aggregate([
  { $match: { createdAt: { $gte: new Date("2024-01-01") } } },
  { $group: { _id: "$status", total: { $sum: "$amount" } } }
])

$match vs find：
├── $match在管道中使用
├── 可以与其他阶段组合
├── 支持索引优化
└── 建议放在管道开头

// 按字段分组
db.orders.aggregate([
  {
    $group: {
      _id: "$status",
      count: { $sum: 1 }
    }
  }
])

// 输出
[
  { "_id": "pending", "count": 50 },
  { "_id": "completed", "count": 100 },
  { "_id": "cancelled", "count": 20 }
]

// 多种聚合函数
db.orders.aggregate([
  {
    $group: {
      _id: "$productId",
      totalSales: { $sum: "$amount" },
      avgAmount: { $avg: "$amount" },
      minAmount: { $min: "$amount" },
      maxAmount: { $max: "$amount" },
      firstOrder: { $first: "$createdAt" },
      lastOrder: { $last: "$createdAt" },
      orderCount: { $sum: 1 },
      uniqueUsers: { $addToSet: "$userId" }
    }
  }
])

// 按多个字段分组
db.orders.aggregate([
  {
    $group: {
      _id: {
        year: { $year: "$createdAt" },
        month: { $month: "$createdAt" },
        status: "$status"
      },
      total: { $sum: "$amount" }
    }
  }
])

// 使用$match筛选分组结果
db.orders.aggregate([
  {
    $group: {
      _id: "$productId",
      totalSales: { $sum: "$amount" }
    }
  },
  {
    $match: {
      totalSales: { $gte: 10000 }
    }
  }
])

// 选择和重命名字段
db.users.aggregate([
  {
    $project: {
      _id: 0,
      name: 1,
      email: 1,
      username: "$name"  // 重命名
    }
  }
])

// 添加计算字段
db.orders.aggregate([
  {
    $project: {
      orderId: "$_id",
      total: { $multiply: ["$price", "$quantity"] },
      discount: { $cond: {
        if: { $gte: ["$quantity", 10] },
        then: 0.1,
        else: 0
      }}
    }
  }
])

// 投影嵌套字段
db.users.aggregate([
  {
    $project: {
      name: 1,
      city: "$address.city",
      country: "$address.country"
    }
  }
])

// 升序排序
db.orders.aggregate([
  { $sort: { createdAt: 1 } }
])

// 降序排序
db.orders.aggregate([
  { $sort: { amount: -1 } }
])

// 多字段排序
db.orders.aggregate([
  { $sort: { status: 1, createdAt: -1 } }
])

// 允许使用磁盘排序
db.orders.aggregate([
  { $sort: { amount: -1 } }
], { allowDiskUse: true })

// 展开数组
db.orders.aggregate([
  { $unwind: "$items" }
])

// 输入
{ "_id": 1, "items": ["A", "B", "C"] }

// 输出
{ "_id": 1, "items": "A" }
{ "_id": 1, "items": "B" }
{ "_id": 1, "items": "C" }

// 保留空数组文档
db.orders.aggregate([
  {
    $unwind: {
      path: "$items",
      preserveNullAndEmptyArrays: true
    }
  }
])

// 展开嵌套对象数组
db.orders.aggregate([
  { $unwind: "$items" },
  {
    $project: {
      productId: "$items.productId",
      quantity: "$items.quantity",
      price: "$items.price"
    }
  }
])

// 左连接
db.orders.aggregate([
  {
    $lookup: {
      from: "users",
      localField: "userId",
      foreignField: "_id",
      as: "user"
    }
  }
])

// 使用管道进行复杂关联
db.orders.aggregate([
  {
    $lookup: {
      from: "products",
      let: { productId: "$productId" },
      pipeline: [
        {
          $match: {
            $expr: { $eq: ["$_id", "$$productId"] }
          }
        },
        { $project: { name: 1, price: 1 } }
      ],
      as: "product"
    }
  }
])

// 关联多个集合
db.orders.aggregate([
  {
    $lookup: {
      from: "users",
      localField: "userId",
      foreignField: "_id",
      as: "user"
    }
  },
  {
    $lookup: {
      from: "products",
      localField: "productId",
      foreignField: "_id",
      as: "product"
    }
  },
  {
    $unwind: "$user"
  },
  {
    $unwind: "$product"
  }
])

// 添加新字段
db.orders.aggregate([
  {
    $addFields: {
      totalAmount: { $multiply: ["$price", "$quantity"] },
      orderYear: { $year: "$createdAt" }
    }
  }
])

$addFields vs $project：
├── $addFields保留所有字段
├── $project需要指定保留字段
├── $addFields可多次使用
└── 都可以添加计算字段

// 并行执行多个聚合管道
db.products.aggregate([
  {
    $facet: {
      categories: [
        { $group: { _id: "$category", count: { $sum: 1 } } }
      ],
      priceRanges: [
        {
          $bucket: {
            groupBy: "$price",
            boundaries: [0, 100, 500, 1000, 5000],
            default: "other"
          }
        }
      ],
      topRated: [
        { $sort: { rating: -1 } },
        { $limit: 5 }
      ],
      totalCount: [
        { $count: "count" }
      ]
    }
  }
])

// 同时获取数据和总数
db.products.aggregate([
  { $match: { status: "active" } },
  {
    $facet: {
      data: [
        { $sort: { createdAt: -1 } },
        { $skip: 0 },
        { $limit: 10 }
      ],
      totalCount: [
        { $count: "count" }
      ]
    }
  }
])

// 算术运算
db.orders.aggregate([
  {
    $project: {
      total: { $add: ["$price", "$tax"] },
      discount: { $subtract: ["$price", "$discount"] },
      quantity: { $multiply: ["$price", "$quantity"] },
      average: { $divide: ["$total", "$count"] },
      remainder: { $mod: ["$total", 100] }
    }
  }
])

// 字符串操作
db.users.aggregate([
  {
    $project: {
      upperName: { $toUpper: "$name" },
      lowerEmail: { $toLower: "$email" },
      fullName: { $concat: ["$firstName", " ", "$lastName"] },
      substring: { $substr: ["$phone", 0, 3] },
      strLength: { $strLenCP: "$name" }
    }
  }
])

// 日期操作
db.orders.aggregate([
  {
    $project: {
      year: { $year: "$createdAt" },
      month: { $month: "$createdAt" },
      day: { $dayOfMonth: "$createdAt" },
      hour: { $hour: "$createdAt" },
      dayOfWeek: { $dayOfWeek: "$createdAt" },
      week: { $week: "$createdAt" },
      formatted: {
        $dateToString: {
          format: "%Y-%m-%d %H:%M:%S",
          date: "$createdAt"
        }
      }
    }
  }
])

// 条件表达式
db.orders.aggregate([
  {
    $project: {
      status: {
        $cond: {
          if: { $gte: ["$amount", 1000] },
          then: "high",
          else: "normal"
        }
      },
      level: {
        $switch: {
          branches: [
            { case: { $lt: ["$amount", 100] }, then: "low" },
            { case: { $lt: ["$amount", 500] }, then: "medium" },
            { case: { $gte: ["$amount", 500] }, then: "high" }
          ],
          default: "unknown"
        }
      }
    }
  }
])

// 数组操作
db.orders.aggregate([
  {
    $project: {
      arraySize: { $size: "$items" },
      firstItem: { $arrayElemAt: ["$items", 0] },
      lastItem: { $arrayElemAt: ["$items", -1] },
      slice: { $slice: ["$items", 2, 3] },
      concatArrays: { $concatArrays: ["$items", "$extras"] },
      filterArray: {
        $filter: {
          input: "$items",
          as: "item",
          cond: { $gte: ["$$item.price", 100] }
        }
      }
    }
  }
])

// 月度销售报表
db.orders.aggregate([
  {
    $match: {
      createdAt: {
        $gte: new Date("2024-01-01"),
        $lt: new Date("2024-02-01")
      }
    }
  },
  {
    $group: {
      _id: {
        year: { $year: "$createdAt" },
        month: { $month: "$createdAt" },
        day: { $dayOfMonth: "$createdAt" }
      },
      totalSales: { $sum: "$amount" },
      orderCount: { $sum: 1 },
      avgOrderValue: { $avg: "$amount" },
      maxOrder: { $max: "$amount" },
      minOrder: { $min: "$amount" }
    }
  },
  { $sort: { "_id.day": 1 } }
])

// 用户行为统计
db.userActions.aggregate([
  {
    $match: {
      timestamp: {
        $gte: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000)
      }
    }
  },
  {
    $group: {
      _id: {
        userId: "$userId",
        action: "$action"
      },
      count: { $sum: 1 },
      lastAction: { $max: "$timestamp" }
    }
  },
  {
    $group: {
      _id: "$_id.userId",
      actions: {
        $push: {
          action: "$_id.action",
          count: "$count",
          lastAction: "$lastAction"
        }
      },
      totalActions: { $sum: "$count" }
    }
  },
  { $sort: { totalActions: -1 } },
  { $limit: 100 }
])

// 库存预警分析
db.products.aggregate([
  {
    $lookup: {
      from: "inventory",
      localField: "_id",
      foreignField: "productId",
      as: "inventory"
    }
  },
  { $unwind: "$inventory" },
  {
    $addFields: {
      stockRatio: {
        $divide: ["$inventory.quantity", "$inventory.threshold"]
      }
    }
  },
  {
    $match: {
      stockRatio: { $lt: 1.2 }
    }
  },
  {
    $project: {
      name: 1,
      currentStock: "$inventory.quantity",
      threshold: "$inventory.threshold",
      stockRatio: 1,
      status: {
        $cond: {
          if: { $lt: ["$stockRatio", 0.5] },
          then: "critical",
          else: "warning"
        }
      }
    }
  },
  { $sort: { stockRatio: 1 } }
])

聚合优化：
├── 尽早使用$match减少数据量
├── 在$match后使用索引
├── 合理使用$project减少字段
├── 避免不必要的$unwind
├── 使用allowDiskUse处理大数据
└── 监控聚合性能

// 允许使用磁盘
db.orders.aggregate([...], { allowDiskUse: true })

// 设置批量大小
db.orders.aggregate([...], { cursor: { batchSize: 100 } })

聚合最佳实践：
├── 理解数据流向
├── 优化管道顺序
├── 合理使用索引
├── 控制内存使用
├── 测试和监控性能
└── 处理大数据使用allowDiskUse