Aggregation

The Aggregates class provides static factory methods that build aggregation pipeline stages. Each method returns an instance of the Bson type, which can in turn be passed to the aggregate method of MongoCollection.

For brevity, you may choose to import the methods of the Aggregates class statically:

import org.mongodb.scala.model.Aggregates._

All the examples below assume this static import.

Match

The $match pipeline stage passes all documents matching the specified filter to the next stage. Though the filter can be an instance of any class that implements Bson, it’s convenient to combine with use of the Filters class. In the example below, it’s assumed that the equal method of the Filters class has been statically imported.

This example creates a pipeline stage that matches all documents where the author field is equal to "Dave":

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

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

Project

The $project pipeline stage passes the projected fields of all documents to the next stage. Though the projection can be an instance of any class that implements Bson, it’s convenient to combine with use of the Projections class. In the example below, it’s assumed that the include, excludeId, and fields methods of the Projections class have been statically imported.

This example creates a pipeline stage that excludes the _id field but includes the title and author fields:

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

Projecting Computed Fields

The $project stage can project computed fields as well.

This example simply projects the qty field into a new field called quantity. In other words, it renames the field:

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

Sample

The $sample pipeline stage randomly select N documents from its input. This example creates a pipeline stage that randomly selects 5 documents from the collection:

sample(5)

Sort

The $sort pipeline stage passes all documents to the next stage, sorted by the specified sort criteria. Though the sort criteria can be an instance of any class that implements Bson, it’s convenient to combine with use of the Sorts class. In the example below, it’s assumed that the descending, ascending, and orderBy methods of the Sorts class have been statically imported.

This example creates a pipeline stage that sorts in descending order according to the value of the age field and then in ascending order according to the value of the posts field:

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

Skip

The $skip pipeline stage skips over the specified number of documents that pass into the stage and passes the remaining documents to the next stage.

This example skips the first 5 documents:

skip(5)

Limit

The $limit pipeline stage limits the number of documents passed to the next stage.

This example limits the number of documents to 10:

limit(10)

Lookup

Starting in 3.2, MongoDB provides a new $lookup pipeline stage that performs a left outer join with another collection to filter in documents from the joined collection for processing.

This example performs a left outer join on the fromCollection collection, joining the local field to the from field and outputted in the joinedOutput field:

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

Group

The $group pipeline stage groups documents by some specified expression and outputs to the next stage a document for each distinct grouping. A group consists of an _id which specifies the expression on which to group, and zero or more accumulators which are evaluated for each grouping. To simplify the expression of accumulators, the driver includes an Accumulators singleton object with factory methods for each of the supported accumulators. In the example below, it’s assumed that the sum and avg methods of the Accumulators class have been statically imported.

This example groups documents by the value of the customerId field, and for each group accumulates the sum and average of the values of the quantity field into the totalQuantity and averageQuantity fields, respectively.

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

Unwind

The $unwind pipeline stage deconstructs an array field from the input documents to output a document for each element.

This example outputs, for each document, a document for each element in the sizes array:

unwind("$sizes")

Available with MongoDB 3.2, this example also includes any documents that have missing or null values for the $sizes field or where the $sizes list is empty:

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

Available with MongoDB 3.2, this example unwinds the sizes array and also outputs the array index into the $position field:

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

Out

The $out pipeline stage outputs all documents to the specified collection. It must be the last stage in any aggregate pipeline:

This example writes the pipeline to the authors collection:

out("authors")

SetWindowFields

The $setWindowFields pipeline stage allows using window operators. This stage partitions the input documents similarly to the $group pipeline stage, optionally sorts them, computes fields in the documents by computing window functions over windows specified per function (a window is a subset of a partition), and outputs the documents. The important difference from the $group pipeline stage is that documents belonging to the same partition or window are not folded into a single document.

The driver includes the WindowedComputations singleton object with factory methods for supported window operators.

This example computes the accumulated rainfall and the average temperature over the past month per each locality from more fine-grained measurements presented via the rainfall and temperature fields:

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

Creating a Pipeline

The above pipeline operators are typically combined into a list and passed to the aggregate method of a MongoCollection. For instance:

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