聚合框架

先决条件

您必须设置以下组件才能运行本指南中的代码示例：

• 一个 test.restaurants集合，其中填充了来自文档资产Github中 restaurants.json文件的文档。

• 以下 import 语句：

import org.mongodb.scala._
import org.mongodb.scala.model.Aggregates._
import org.mongodb.scala.model.Accumulators._
import org.mongodb.scala.model.Filters._
import org.mongodb.scala.model.Projections._

连接到 MongoDB 部署

首先，连接到 MongoDB 部署，然后声明并定义MongoDatabase和MongoCollection实例。

以下代码连接到在端口27017上的localhost上运行的独立 MongoDB 部署。 然后，定义database变量以引用test数据库，并collection变量以引用restaurants集合：

val mongoClient: MongoClient = MongoClient()
val database: MongoDatabase = mongoClient.getDatabase("test")
val collection: MongoCollection[Document] = database.getCollection("restaurants")

要了解有关连接到 MongoDB 部署的更多信息，请参阅连接到 MongoDB教程。

执行聚合

要执行聚合，请将聚合阶段列表传递给 MongoCollection.aggregate() 方法。驱动程序提供了 Aggregates 辅助工具类，其中包含聚合阶段的构建者。

在此示例中，聚合管道执行以下任务：

• 使用$match阶段筛选categories数组字段包含元素"Bakery"的文档。 该示例使用Aggregates.filter()构建$match阶段。

• 使用 $group 阶段按 stars字段对匹配文档进行群组，并累积每个不同的 stars 值的文档计数。该示例使用 Aggregates.group()构建$group 阶段，并使用 Accumulators.sum()构建累加器表达式。对于在 $group 阶段使用的累加器表达式，驱动程序提供了 Accumulators 辅助工具类。

collection.aggregate(Seq(
    Aggregates.filter(Filters.equal("categories", "Bakery")),
    Aggregates.group("$stars", Accumulators.sum("count", 1))
)).printResults()

collection.aggregate(
  Seq(
    Aggregates.project(
      Projections.fields(
        Projections.excludeId(),
        Projections.include("name"),
        Projections.computed(
          "firstCategory",
          Document("$arrayElemAt"-> Seq("$categories", 0))
        )
      )
    )
  )
).printResults()

collection.aggregate(
  Seq(Aggregates.filter(Filters.eq("categories", "Bakery")),
      Aggregates.group("$stars", Accumulators.sum("count", 1)))
).explain().printResults()