Spring Data MongoDB 聚合框架：实现复杂分组、统计与输出扁平化

本文详细介绍了如何将复杂的 MongoDB 聚合查询转换为 Spring Data MongoDB 的 Java 代码。具体地，我们将一个按年份和状态分组、统计计数，并最终扁平化输出结果的 MongoDB 聚合管道，通过 Aggregation 框架中的 project、group、replaceWith 和 unset 等操作符，逐步构建出功能完备的 Java 实现。

在现代应用程序开发中，MongoDB 聚合框架是处理和转换集合数据的强大工具。它允许开发者构建复杂的数据管道，对文档进行过滤、分组、转换和计算。当我们需要在 Java 应用程序中利用 Spring Data MongoDB 执行这些复杂的聚合查询时，理解如何将 MongoDB 原生语法映射到 Spring Data MongoDB 的 Aggregation 框架至关重要。

MongoDB 聚合查询解析

首先，我们来分析一个典型的 MongoDB 聚合查询，该查询旨在按年份和状态对文档进行分组，统计每个分组的文档数量，并最终将结果扁平化，使其更易于消费。

db.collection.aggregate([
    {
        $group: {
            _id: {
                year: { $year: "$createdAt" },
                status: "$status"
            },
            count: { $sum: 1 }
        }
    },
    { $replaceWith: { $mergeObjects: [ "$_id", "$$ROOT" ] } },
    { $unset: "_id" }
])

这个聚合管道包含三个主要阶段：

1. $group 阶段：
   • _id 字段定义了分组键，这里是根据 createdAt 字段的年份 ($year) 和 status 字段进行组合分组。
   • count: { $sum: 1 } 用于计算每个分组中的文档数量。
2. $replaceWith 阶段：
   • 此操作将当前文档完全替换为指定的内容。在这里，它使用 $mergeObjects 将 $ROOT（当前文档，此时包含 _id 和 count）与 $_id（分组键）合并。其目的是将 _id 中的 year 和 status 字段提升到文档的顶层，同时保留 count 字段。
3. $unset 阶段：
   • 此操作用于从文档中移除指定的字段。在这里，它移除了 _id 字段，因为其内容已经通过 $replaceWith 提升到顶层。

最终，查询将返回一个类似 { "year": 2023, "status": "active", "count": 10 } 的扁平化结构。

Spring Data MongoDB 聚合框架实现

将上述复杂的 MongoDB 聚合查询转换为 Spring Data MongoDB 的 Java 代码，需要利用 Aggregation 类及其提供的各种操作符。以下是分步实现过程。

1. 数据投影 ($project)

在进行分组之前，我们需要从 createdAt 字段中提取年份。虽然 $group 阶段的 _id 中可以直接使用 $year 操作符，但为了代码的清晰度和模块化，我们也可以选择在 project 阶段提前处理。然而，在本例中，更直接且符合 MongoDB 原生 $group 语义的方式是先投影出必要的字段，或者直接在 $group 的 _id 中使用日期操作符。为了更好地映射到 Aggregation.group 的 Fields.from 结构，我们可以在 $project 阶段显式地将年份和状态作为独立字段准备好。

import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.DateOperators;
import org.springframework.data.mongodb.core.aggregation.ProjectionOperation;

// 阶段一：投影操作，提取年份和保留状态
ProjectionOperation projectOperation = Aggregation.project("status")
        .and(DateOperators.Year.yearOf("createdAt")).as("year");

这里，我们投影了 status 字段，并使用 DateOperators.Year.yearOf("createdAt") 提取 createdAt 字段的年份，将其命名为 year。

2. 分组与计数 ($group)

接下来是核心的分组操作。我们需要根据上一步投影出的 year 和 status 字段进行分组，并计算每个分组的文档数量。

import org.springframework.data.mongodb.core.aggregation.Fields;
import org.springframework.data.mongodb.core.aggregation.GroupOperation;

// 阶段二：分组操作，按年份和状态分组并计数
GroupOperation groupOperation = Aggregation.group(
            Fields.from(
                Fields.field("year", "year"), // 分组键：使用投影出的 year 字段
                Fields.field("status", "status") // 分组键：使用投影出的 status 字段
            )
        ).count().as("count"); // 计算每个分组的文档数量，并命名为 count

Aggregation.group(Fields.from(...)) 允许我们定义一个复合分组键，这里我们指定了 year 和 status 作为分组依据。count().as("count") 等价于 MongoDB 的 $sum: 1。

3. 替换与合并 ($replaceWith)

$replaceWith 操作在 Spring Data MongoDB 中由 ReplaceWithOperation 实现。它通常与 ObjectOperators.MergeObjects 结合使用，以模拟 MongoDB 原生 $mergeObjects 的行为。

百度智能云·曦灵

百度旗下的AI数字人平台

下载

import org.springframework.data.mongodb.core.aggregation.ReplaceWithOperation;
import org.springframework.data.mongodb.core.aggregation.ObjectOperators;

// 阶段三：替换操作，将 _id 内容提升到顶层
ReplaceWithOperation replaceWithOperation = ReplaceWithOperation.replaceWithValueOf(
        ObjectOperators.MergeObjects.mergeValuesOf("$_id").mergeWith("$$ROOT")
);

mergeValuesOf("$_id") 表示获取当前文档的 _id 字段的值（即分组后的 year 和 status），然后 mergeWith("$$ROOT") 将其与当前文档的其余部分（此时包含 _id 和 count）合并。由于 _id 包含了 year 和 status，合并后这些字段会被提升。

4. 移除字段 ($unset)

最后，我们使用 UnsetOperation 来移除不再需要的 _id 字段。

import org.springframework.data.mongodb.core.aggregation.UnsetOperation;

// 阶段四：移除 _id 字段
UnsetOperation unsetOperation = UnsetOperation.unset("_id");

完整 Java 聚合代码示例

将上述所有阶段组合起来，形成一个完整的 Aggregation 管道，并通过 MongoOperations 执行：

import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.AggregationResults;
import org.springframework.data.mongodb.core.aggregation.DateOperators;
import org.springframework.data.mongodb.core.aggregation.Fields;
import org.springframework.data.mongodb.core.aggregation.ObjectOperators;
import org.springframework.data.mongodb.core.aggregation.ProjectionOperation;
import org.springframework.data.mongodb.core.aggregation.GroupOperation;
import org.springframework.data.mongodb.core.aggregation.ReplaceWithOperation;
import org.springframework.data.mongodb.core.aggregation.UnsetOperation;
import org.springframework.stereotype.Service;

// 假设您已经注入了 MongoOperations
@Service
public class AggregationService {

    private final MongoOperations mongoOperations;

    public AggregationService(MongoOperations mongoOperations) {
        this.mongoOperations = mongoOperations;
    }

    public AggregationResults