聚合

匹配

$match管道阶段将与指定过滤匹配的所有文档传递到下一阶段。 虽然过滤可以是实现Bson的任何类的实例，但使用筛选器类中的方法会更方便。

以下示例创建了一个管道阶段，用于匹配author字段值为"Dave"的所有文档：

`match`(equal("author", "Dave"))

filter(equal("author", "Dave"))

项目

$project管道阶段将所有文档的投影字段传递到下一阶段。 虽然投影可以是任何实现Bson的类的实例，但使用Projections类中的方法会更方便。

以下示例创建一个管道阶段，其中不包括 _id 字段，但包括 title 和 author 字段：

project(fields(include("title", "author"), excludeId()))

project(computed("quantity", "$qty"))

抽检 (Sample)

$sample管道阶段从输入文档中随机选择N文档。 以下示例使用sample()方法从集合中随机选择5文档：

sample(5)

Sort

$sort管道阶段将所有文档传递到下一阶段，并按指定的排序条件进行排序。 虽然排序条件可以是任何实现Bson的类的实例，但使用Sorts类中的方法会更方便。

以下示例创建了管道阶段，根据 age 字段的值以降序排序，然后根据 posts 字段的值以升序排序：

sort(orderBy(descending("age"), ascending("posts")))

跳过

$skip管道阶段会跳过进入该阶段的指定数量的文档，并将剩余文档传递到下一阶段。

以下示例会跳过前5个文档：

skip(5)

Limit

$limit管道阶段限制传递到下一阶段的文档数量。

以下示例将文档数量限制为10 ：

limit(10)

Lookup

$lookup管道阶段与另一个集合执行左外连接，以过滤已连接集合中的文档进行处理。

以下示例对fromCollection集合执行左外连接，将local字段连接到from字段，并在joinedOutput字段中输出：

lookup("fromCollection", "local", "from", "joinedOutput")

GROUP

$group管道阶段按某些指定的表达式对文档进行分组，并将每个不同分组的文档输出到下一阶段。 一个群组由_id和零个或多个累加器组成，其中 指定了要群组的表达式，而零个或多个累加器则针对每个分组进行评估。

为了简化累加器的表达式，驱动程序包含一个Accumulators单例对象，其中包含每个受支持累加器的工厂方法。

以下示例按customerId字段的值对文档进行分组，并将每个群组的数量字段值的总和和平均值分别累加到totalQuantity和averageQuantity字段中：

group("$customerId", sum("totalQuantity", "$quantity"), avg("averageQuantity", "$quantity"))

Unwind

$unwind管道阶段解构输入文档中的大量字段，为每个元素输出一个文档。

以下示例为每个文档输出sizes数组中每个元素的文档：

unwind("$sizes")

以下示例还包括sizes字段缺失值或 null 值，或者sizes列表为空的任何文档：

unwind("$sizes", UnwindOptions().preserveNullAndEmptyArrays(true))

以下示例展开sizes数组，并将数组索引输出到position字段中：

unwind("$sizes", UnwindOptions().includeArrayIndex("$position"))

SetWindowFields

$setWindowFields管道阶段允许使用窗口操作符。 此阶段对输入文档进行分区，类似于 $group管道阶段，可以选择对它们进行排序，通过在每个函数指定的Windows上计算窗口函数来计算文档中的字段，然后输出文档。 窗口是分区的子集。

与$group管道阶段的重要区别在于，属于同一分区或窗口的文档不会折叠为单个文档。

驾驶员包含WindowedComputations单例对象，其中包含受支持窗口操作符的工厂方法。

以下示例根据rainfall和temperature字段中提供的更细粒度的测量值，计算每个地点过去一个月的累计降雨量和平均气温：

val pastMonth: Window = Windows.timeRange(-1, MongoTimeUnit.MONTH, Windows.Bound.CURRENT)
setWindowFields(Some("$localityId"), Some(Sorts.ascending("measurementDateTime")),
  WindowedComputations.sum("monthlyRainfall", "$rainfall", Some(pastMonth)),
  WindowedComputations.avg("monthlyAvgTemp", "$temperature", Some(pastMonth)))

组装管道

管道操作符通常组合成一个列表并传递给MongoCollection的aggregate()方法：

collection.aggregate(List(filter(equal("author", "Dave")),
    group("$customerId", sum("totalQuantity", "$quantity"),
                         avg("averageQuantity", "$quantity")),
    out("authors")))