【Spring连载】使用Spring Data访问 MongoDB----Aggregation Framework支持

一、基础槪念

Spring Data MongoDB中的聚合框架支持基于以下关键抽象：Aggregation、AggregationDefinition和AggregationResults。

• 聚合Aggregation
  Aggregation表示MongoDB聚合操作，并保存聚合管道（aggregation pipeline）指令的描述。聚合是通过调用Aggregation类的相应newAggregation(…)静态工厂方法创建的，该方法采用AggregateOperation列表和一个可选输入类。
  实际的聚合操作由MongoTemplate的聚合方法运行，该方法将所需的输出类作为参数。
• 类型聚合TypedAggregation
  TypedAggregation与Aggregation一样，包含聚合管道的指令和对输入类型的引用，用于将域属性映射到实际document字段。
  在运行时，考虑到潜在的@Field注解，根据给定的输入类型检查字段引用。

在3.2版本中做出更改，引用不存在的属性不再引发错误。要恢复以前的行为，请使用AggregationOptions的strictMapping选项。

• 聚合定义AggregationDefinition
  AggregationDefinition表示MongoDB聚合管道操作，并描述该聚合步骤中应执行的处理。尽管你可以手动创建AggregationDefinition，但建议使用Aggregate类提供的静态工厂方法来构造AggregateOperation。
• 聚合结果AggregationResults
  AggregationResults是聚合操作结果的容器。它以Document的形式提供对原始聚合结果的访问，以访问映射的对象和有关聚合的其他信息。

以下列表展示了使用Spring Data MongoDB对MongoDB聚合框架的支持的典型示例：

java">Aggregation agg = newAggregation(
    pipelineOP1(),
    pipelineOP2(),
    pipelineOPn()
);

AggregationResults<OutputType> results = mongoTemplate.aggregate(agg, "INPUT_COLLECTION_NAME", OutputType.class);
List<OutputType> mappedResult = results.getMappedResults();

注意，如果你提供一个输入类作为newAggregation方法的第一个参数，那么MongoTemplate将从该类派生输入集合的名称。否则，如果不指定输入类，则必须显式地提供输入集合的名称。如果同时提供了输入类和输入集合，则后者优先。
支持的聚合操作和阶段
MongoDB聚合框架提供了以下类型的聚合阶段和操作:

• addFields - AddFieldsOperation
• bucket / bucketAuto - BucketOperation / BucketAutoOperation
• count - CountOperation
• densify - DensifyOperation
• facet - FacetOperation
• geoNear - GeoNearOperation
• graphLookup - GraphLookupOperation
• group - GroupOperation
• limit - LimitOperation
• lookup - LookupOperation
• match - MatchOperation
• merge - MergeOperation
• project - ProjectionOperation
• redact - RedactOperation
• replaceRoot - ReplaceRootOperation
• sample - SampleOperation
• set - SetOperation
• setWindowFields - SetWindowFieldsOperation
• skip - SkipOperation
• sort / sortByCount - SortOperation / SortByCountOperation
• unionWith - UnionWithOperation
• unset - UnsetOperation
• unwind - UnwindOperation

不支持的聚合阶段(如MongoDB Atlas的$search)可以通过实现AggregationOperation或Bson表示来提供。Aggregation.stage是通过提供其JSON来注册pipeline stage的快捷方式。

Aggregation.stage("""
    { $search : {
        "near": {
          "path": "released",
          "origin": { "$date": { "$numberLong": "..." } } ,
          "pivot": 7
        }
      }
    }
""");

在撰写本文时，Spring Data MongoDB为以下聚合运算符提供支持：
表1：Spring Data MongoDB当前支持的聚合运算符

*操作由Spring Data MongoDB映射或添加。
请注意，此处未列出的聚合操作目前不被Spring Data MongoDB支持。比较聚合运算符表示为Criteria表达式。

二、投影表达式Projection Expressions

投影表达式用于定义特定聚合步骤的结果字段。投影表达式可以通过Aggregation类的project方法定义，也可以通过传递String对象列表或聚合框架Fields对象来定义。可以使用and(String)方法通过fluent API使用附加字段扩展投影，并使用as(String)方法对其起别名。请注意，你还可以使用聚合框架的Fields.field静态工厂方法定义具有别名的字段，然后可以使用该方法构造新的Fields实例。在后面的聚合阶段中对投影字段的引用仅对包含字段的字段名或其别名（包括新定义的字段及其别名）有效。未包含在投影中的字段不能在以后的聚合阶段中引用。以下列表展示了投影表达式的示例：
例1：投影表达式示例

// generates {$project: {name: 1, ***Price: 1}}
project("name", "***Price")

// generates {$project: {thing1: $thing2}}
project().and("thing1").as("thing2")

// generates {$project: {a: 1, b: 1, thing2: $thing1}}
project("a","b").and("thing1").as("thing2")

例2。使用投影和排序的多阶段聚合

// generates {$project: {name: 1, ***Price: 1}}, {$sort: {name: 1}}
project("name", "***Price"), sort(ASC, "name")

// generates {$project: {name: $firstname}}, {$sort: {name: 1}}
project().and("firstname").as("name"), sort(ASC, "name")

// does not work
project().and("firstname").as("name"), sort(ASC, "firstname")

项目操作的更多示例可以在AggregationTests类中找到。请注意，关于投影表达式的更多细节可以在MongoDB Aggregation Framework参考文档的相应部分中找到。

三、分面分类法Faceted Classification

从3.4版本开始，MongoDB使用聚合框架支持分面分类(faceted classification)。分面分类使用语义类别（通用或特定主题），这些类别组合在一起创建完整的分类条目。流经聚合管道的Documents被分类到存储桶中。多方面分类允许对同一组输入documents进行各种聚合，而无需多次检索输入documents。

3.1 桶Buckets

Bucket操作根据指定的表达式和bucket边界将传入documents分类为组，称为buckets。Bucket操作需要分组字段或分组表达式。你可以使用Aggregate类的bucket()和bucketAuto()方法来定义它们。BucketOperation和BucketAutoOperation可以公开基于输入文档的聚合表达式的累积（a***umulations）。你可以使用with…()方法和andOutput(String)方法，通过fluent API使用附加参数来扩展bucket操作。你可以使用as(String)方法对操作起别名。每个bucket在输出中表示为一个document。
BucketOperation采用一组定义的边界将传入documents分组到这些类别中。需要对边界进行排序。以下列表展示了bucket操作的一些示例：
例3：Bucket操作示例

// generates {$bucket: {groupBy: $price, boundaries: [0, 100, 400]}}
bucket("price").withBoundaries(0, 100, 400);

// generates {$bucket: {groupBy: $price, default: "Other" boundaries: [0, 100]}}
bucket("price").withBoundaries(0, 100).withDefault("Other");

// generates {$bucket: {groupBy: $price, boundaries: [0, 100], output: { count: { $sum: 1}}}}
bucket("price").withBoundaries(0, 100).andOutputCount().as("count");

// generates {$bucket: {groupBy: $price, boundaries: [0, 100], 5, output: { titles: { $push: "$title"}}}
bucket("price").withBoundaries(0, 100).andOutput("title").push().as("titles");

BucketAutoOperation确定边界，试图将documents平均分配到指定数量的存储桶中。BucketAutoOperation可选地采用一个粒度值，该值指定要使用的首选数字序列，以确保计算的边界边以首选整数或10的幂结束。以下列表展示了bucket操作的示例：
例4：Bucket操作示例

// generates {$bucketAuto: {groupBy: $price, buckets: 5}}
bucketAuto("price", 5)

// generates {$bucketAuto: {groupBy: $price, buckets: 5, granularity: "E24"}}
bucketAuto("price", 5).withGranularity(Granularities.E24).withDefault("Other");

// generates {$bucketAuto: {groupBy: $price, buckets: 5, output: { titles: { $push: "$title"}}}
bucketAuto("price", 5).andOutput("title").push().as("titles");

要在buckets中创建输出字段，bucket操作可以通过andOutput()使用AggregationExpression，通过andOutputExpression()使用SpEL表达式。
请注意，有关bucket表达式的更多详细信息可以在MongoDB Aggregation Framework参考文档的$ bucket部分和$bucketAuto部分找到。

3.2 多方面的聚合Multi-faceted Aggregation

多个聚合管道可用于创建多方面聚合，这些聚合在单个聚合阶段内表征多个维度（或方面）的数据。多面聚合提供了多个过滤器和分类，以指导数据浏览和分析。faceting的一个常见实现是，有多少在线零售商通过对产品价格、制造商、尺寸和其他因素应用过滤器来缩小搜索结果的范围。
你可以使用Aggregation类的facet()方法定义FacetOperation。你还可以使用and()方法使用多个聚合管道对其进行自定义。每个子管道在输出document中都有自己的字段，其结果存储为documents数组。
子管道可以在分组之前投影和过滤输入documents。常见的用例包括在分类之前提取日期部分或进行计算。以下列表展示了方面（facet）操作示例：
例5：Facet操作示例

// generates {$facet: {categorizedByPrice: [ { $match: { price: {$exists : true}}}, { $bucketAuto: {groupBy: $price, buckets: 5}}]}}
facet(match(Criteria.where("price").exists(true)), bucketAuto("price", 5)).as("categorizedByPrice"))

// generates {$facet: {categorizedByCountry: [ { $match: { country: {$exists : true}}}, { $sortByCount: "$country"}]}}
facet(match(Criteria.where("country").exists(true)), sortByCount("country")).as("categorizedByCountry"))

// generates {$facet: {categorizedByYear: [
//     { $project: { title: 1, publicationYear: { $year: "publicationDate"}}},
//     { $bucketAuto: {groupBy: $price, buckets: 5, output: { titles: {$push:"$title"}}}
// ]}}
facet(project("title").and("publicationDate").extractYear().as("publicationYear"),
      bucketAuto("publicationYear", 5).andOutput("title").push().as("titles"))
  .as("categorizedByYear"))

请注意，关于facet操作的更多细节可以在MongoDB Aggregation Framework参考文档的$facet部分中找到。

3.3 按计数排序Sort By Count

按计数排序操作根据指定表达式的值对传入documents进行分组，计算每个distinct组中的documents数，并按计数对结果进行排序。当使用分面分类时，它提供了一个方便的快捷方式来应用排序。按计数排序操作需要分组字段或分组表达式。以下列表展示了按计数排序的示例：
例6：按计数排序示例

// generates { $sortByCount: "$country" }
sortByCount("country");

按计数排序操作相当于以下BSON(二进制JSON):

{ $group: { _id: <expression>, count: { $sum: 1 } } },
{ $sort: { count: -1 } }

3.4 投影表达式中的Spring表达式支持

通过ProjectionOperation 和BucketOperation 类的andExpression方法，框架支持在投影表达式中使用SpEL表达式。该特性允许你将所需的表达式定义为SpEL表达式。在运行查询时，SpEL表达式被转换为对应的MongoDB投影表达式部分。这种安排使表达复杂的计算变得容易很多。

3.4.1 使用SpEL表达式的复杂计算

参见下面的SpEL表达式:

1 + (q + 1) / (q - 1)

将上述表达式翻译成以下投影表达式部分:

{ "$add" : [ 1, {
    "$divide" : [ {
        "$add":["$q", 1]}, {
        "$subtract":[ "$q", 1]}
    ]
}]}

你可以在聚合框架例5和聚合框架例6中看到更多上下文中的示例。你可以在SpelExpressionTransformerUnitTests中找到更多受支持的SpEL表达式构造的用法示例。
支持的SpEL转换

除了上表中展示的转换之外，你还可以使用标准的SpEL操作，例如new来(例如)创建数组并通过其名称引用表达式(后面跟着括号中要使用的参数)。下面的例子展示了如何以这种方式创建一个数组:

// { $setEquals : [$a, [5, 8, 13] ] }
.andExpression("setEquals(a, new int[]{5, 8, 13})");

3.5 聚合框架示例Aggregation Framework Examples

本节中的示例演示了MongoDB Aggregation Framework与Spring Data MongoDB的使用模式。

3.5.1 Aggregation Framework 例1

在这个介绍性示例中，我们希望聚合一个标签列表，以从MongoDB集合（称为tags）中获得特定标签的出现次数，该集合按出现次数降序排序。此示例演示了分组、排序、投影（选择）和展开（结果拆分）的用法。

class TagCount {
 String tag;
 int n;
}

Aggregation agg = newAggregation(
    project("tags"),
    unwind("tags"),
    group("tags").count().as("n"),
    project("n").and("tag").previousOperation(),
    sort(DESC, "n")
);

AggregationResults<TagCount> results = mongoTemplate.aggregate(agg, "tags", TagCount.class);
List<TagCount> tagCount = results.getMappedResults();

前面的列表使用以下算法：

1. 使用newAggregation静态工厂方法创建一个新的聚合，我们将聚合操作列表传递给该方法。这些聚合操作定义了聚合的聚合管道。
2. 使用投影操作从输入集合中选择tags字段（它是一个字符串数组）。
3. 使用展开操作为tags数组中的每个tag生成一个新document。
4. 使用group操作为每个tags值定义一个组，我们将为其聚合出现次数（通过使用count聚合运算符并将结果收集到一个名为n的新字段中）。
5. 选择n字段，并为上一个group操作（因此调用previousOperation()）生成的ID字段创建一个别名，名称为tag。
6. 使用排序操作可以按tags的出现次数降序对结果列表进行排序。
7. 调用MongoTemplate上的聚合方法，让MongoDB执行实际的聚合操作，并将创建的Aggregation作为参数。

请注意，输入集合被显式指定为聚合方法的tags参数。如果没有显式指定输入集合的名称，则它是从作为第一个参数传递给newAggregation方法的输入类派生的。

3.5.2 Aggregation Framework 例2

此示例基于MongoDB聚合框架文档中的各州最大和最小城市示例。这里添加了额外的排序，以在不同的MongoDB版本中产生稳定的结果。在这里，我们想通过使用聚合框架返回每个州按人口划分的最小和最大城市。这个例子演示了分组、排序和投影（选择）。

class ZipInfo {
   String id;
   String city;
   String state;
   @Field("pop") int population;
   @Field("loc") double[] location;
}

class City {
   String name;
   int population;
}

class ZipInfoStats {
   String id;
   String state;
   City biggestCity;
   City smallestCity;
}

TypedAggregation<ZipInfo> aggregation = newAggregation(ZipInfo.class,
    group("state", "city")
       .sum("population").as("pop"),
    sort(ASC, "pop", "state", "city"),
    group("state")
       .last("city").as("biggestCity")
       .last("pop").as("biggestPop")
       .first("city").as("smallestCity")
       .first("pop").as("smallestPop"),
    project()
       .and("state").previousOperation()
       .and("biggestCity")
          .nested(bind("name", "biggestCity").and("population", "biggestPop"))
       .and("smallestCity")
          .nested(bind("name", "smallestCity").and("population", "smallestPop")),
    sort(ASC, "state")
);

AggregationResults<ZipInfoStats> result = mongoTemplate.aggregate(aggregation, ZipInfoStats.class);
ZipInfoStats firstZipInfoStats = result.getMappedResults().get(0);

请注意，ZipInfo类映射给定输入集合的结构。ZipInfoStats类以所需的输出格式定义结构。
前面的列表使用以下算法：

1. 使用group操作可以从输入集合中定义组。分组标准是state和city字段的组合，这形成了组的ID结构。我们使用求和运算符从分组的元素中聚合population属性的值，并将结果保存在pop字段中。
2. 使用排序操作按pop、state和city字段按升序对中间结果进行排序，使最小的城市位于结果的顶部，最大的城市位于其底部。请注意，state和city的排序是根据组ID字段（Spring Data MongoDB处理的）隐式执行的。
3. 再次使用分组操作按state对中间结果进行分组。请注意，state再次隐式引用组ID字段。我们选择最大和最小城市的名称和人口计数，并分别调用project 操作中的last(…)和first(…​) 运算符。
4. 从上一组操作中选择state字段。请注意，state再次隐式引用组ID字段。因为我们不希望隐式生成的ID出现，所以我们使用and(previousOperation()).exclude()从上一个操作中排除该ID。因为我们想在输出类中填充嵌套的City结构，所以我们必须使用嵌套方法发出适当的子documents。
5. 在排序操作中，按state名称的升序对StateStats的结果列表进行排序。

请注意，我们从作为第一个参数传递给newAggregation方法的ZipInfo类派生输入集合的名称。

3.5.3 Aggregation Framework 例3

此示例基于MongoDB聚合框架文档中人口超过1000万的州的示例。我们添加了额外的排序，以在不同的MongoDB版本中产生稳定的结果。在这里，我们想使用聚合框架返回人口超过1000万的所有州。这个例子演示了分组、排序和匹配（过滤）。

class StateStats {
   @Id String id;
   String state;
   @Field("totalPop") int totalPopulation;
}

TypedAggregation<ZipInfo> agg = newAggregation(ZipInfo.class,
    group("state").sum("population").as("totalPop"),
    sort(ASC, previousOperation(), "totalPop"),
    match(where("totalPop").gte(10 * 1000 * 1000))
);

AggregationResults<StateStats> result = mongoTemplate.aggregate(agg, StateStats.class);
List<StateStats> stateStatsList = result.getMappedResults();

前面的列表使用以下算法：

1. 按state字段对输入集合进行分组，计算population字段的总和，并将结果存储在新字段“totalPop”中。
2. 除“totalPop”字段外，还根据上一组操作的id引用按升序对中间结果进行排序。
3. 使用匹配操作筛选中间结果，该操作接受Criteria查询作为参数。

请注意，我们从作为第一个参数传递给newAggregation方法的ZipInfo类派生输入集合的名称。

3.5.4 Aggregation Framework 例4

这个例子演示了在投影运算中使用简单的算术运算。

class Product {
    String id;
    String name;
    double ***Price;
    int spaceUnits;
}

TypedAggregation<Product> agg = newAggregation(Product.class,
    project("name", "***Price")
        .and("***Price").plus(1).as("***PricePlus1")
        .and("***Price").minus(1).as("***PriceMinus1")
        .and("***Price").multiply(1.19).as("grossPrice")
        .and("***Price").divide(2).as("***PriceDiv2")
        .and("spaceUnits").mod(2).as("spaceUnitsMod2")
);

AggregationResults<Document> result = mongoTemplate.aggregate(agg, Document.class);
List<Document> resultList = result.getMappedResults();

注意，我们从作为第一个参数传递给newAggregation方法的Product类中派生输入集合的名称。

3.5.5 Aggregation Framework 例5

这个例子演示了在投影运算中使用源自SpEL表达式的简单算术运算。

class Product {
    String id;
    String name;
    double ***Price;
    int spaceUnits;
}

TypedAggregation<Product> agg = newAggregation(Product.class,
    project("name", "***Price")
        .andExpression("***Price + 1").as("***PricePlus1")
        .andExpression("***Price - 1").as("***PriceMinus1")
        .andExpression("***Price / 2").as("***PriceDiv2")
        .andExpression("***Price * 1.19").as("grossPrice")
        .andExpression("spaceUnits % 2").as("spaceUnitsMod2")
        .andExpression("(***Price * 0.8  + 1.2) * 1.19").as("grossPriceIncludingDiscountAndCharge")

);

AggregationResults<Document> result = mongoTemplate.aggregate(agg, Document.class);
List<Document> resultList = result.getMappedResults();

3.5.6 Aggregation Framework 例6

此示例演示了在投影操作中使用从SpEL表达式派生的复杂算术运算。
注意：传递给addExpression方法的附加参数可以根据其位置与索引器表达式一起引用。在本例中，我们使用[0]引用参数数组的第一个参数。当将SpEL表达式转换为MongoDB聚合框架表达式时，外部参数表达式将替换为它们各自的值。

class Product {
    String id;
    String name;
    double ***Price;
    int spaceUnits;
}

double shippingCosts = 1.2;

TypedAggregation<Product> agg = newAggregation(Product.class,
    project("name", "***Price")
        .andExpression("(***Price * (1-discountRate)  + [0]) * (1+taxRate)", shippingCosts).as("salesPrice")
);

AggregationResults<Document> result = mongoTemplate.aggregate(agg, Document.class);
List<Document> resultList = result.getMappedResults();

注意，我们还可以在SpEL表达式中引用document的其他字段。

3.5.7 Aggregation Framework 例7

这个例子使用了条件投影。它派生自$cond参考文档。

public class InventoryItem {

  @Id int id;
  String item;
  String description;
  int qty;
}

public class InventoryItemProjection {

  @Id int id;
  String item;
  String description;
  int qty;
  int discount
}

TypedAggregation<InventoryItem> agg = newAggregation(InventoryItem.class,
  project("item").and("discount")
    .applyCondition(ConditionalOperator.newBuilder().when(Criteria.where("qty").gte(250))
      .then(30)
      .otherwise(20))
    .and(ifNull("description", "Unspecified")).as("description")
);

AggregationResults<InventoryItemProjection> result = mongoTemplate.aggregate(agg, "inventory", InventoryItemProjection.class);
List<InventoryItemProjection> stateStatsList = result.getMappedResults();

此单步聚合对inventory集合使用投影操作。我们通过对数量大于或等于250的所有库存项目使用条件运算来投影折扣字段。对描述字段执行第二条件投影。我们将Unspecified描述应用于所有没有description字段的项或具有null描述的项。
从MongoDB 3.6开始，可以通过使用条件表达式从投影中排除字段。
例7：条件聚合投影

TypedAggregation<Book> agg = Aggregation.newAggregation(Book.class,
  project("title")
    .and(ConditionalOperators.when(***parisonOperators.valueOf("author.middle") --------1    
        .equalToValue(""))                                                      --------2    
        .then("$$REMOVE")                                                       --------3    
        .otherwiseValueOf("author.middle")                                      --------4    
    )
	.as("author.middle"));
	
1. 如果字段author.middle的值
2. 不包含值，
3. 然后使用$$REMOVE排除该字段。
4. 否则，添加author.middle的字段值。