MongoDB 聚合操作的 Java 实践

1. 概述

本文将深入探讨如何使用 MongoDB Java 驱动 来操作 MongoDB 的 聚合框架（Aggregation Framework）。

我们会先理解聚合的基本概念，然后准备一个真实数据集。最后，通过 Java 代码演示如何使用 Aggregates 构建器 实现常见的聚合操作。

2. 什么是聚合操作？

✅ 聚合（Aggregation）是 MongoDB 中用于数据分析、提取有价值信息的核心机制。

它通过多个处理阶段串联成一条“流水线”（pipeline），前一个阶段的输出自动成为下一个阶段的输入。

以下是常用聚合阶段及其类比（便于理解）：

聚合阶段	类比 SQL	说明
project	SELECT	选择需要的字段，也可计算并添加衍生字段
match	WHERE	按条件过滤文档
group	GROUP BY	按指定规则分组（如计数、求和），每组返回一个文档
sort	ORDER BY	按字段升序或降序排序
count	COUNT	统计文档数量
limit	LIMIT	限制返回结果数量
out	SELECT INTO 新表	将结果写入指定集合（只能作为流水线最后一个阶段）

⚠️ 注意：这里的 SQL 类比仅为帮助理解，实际行为可能略有差异。

接下来我们会用 Java 代码逐一演示这些阶段的使用。在此之前，先准备好数据环境。

3. 数据库准备

3.1. 数据集说明

学习数据库操作，首先得有数据。本文选用一个公开的 RESTful 接口：restcountries.com，它提供全球所有国家的详细信息，格式为 JSON，非常适合做聚合分析。

我们主要使用以下字段：

• name：国家名称，如 United States of America
• alpha3Code：国家三字母代码，如 IND（印度）
• region：所属地理区域，如 Europe
• area：国土面积（平方公里）
• languages：官方语言数组，如 English
• borders：邻国三字母代码数组

目标：将这些数据导入 MongoDB 集合，进行后续分析。

3.2. 导入 MongoDB

步骤如下：

1. 调用接口获取所有国家数据：

   curl https://restcountries.com/v3.1/all > countries.json
   

2. 使用 mongoimport 导入：

   mongoimport --db country_db --collection countries --file countries.json --jsonArray
   

导入成功后，集合中应有约 250 个文档。

4. Java 聚合操作实战

现在，我们用 Java 对 countries 集合进行聚合分析。所有示例基于 JUnit 测试。

数据库连接初始化

@BeforeClass
public static void setUpDB() {
    mongoClient = MongoClients.create("mongodb://localhost:27017");
    database = mongoClient.getDatabase("country_db");
    collection = database.getCollection("countries");
}

为简化代码，建议添加静态导入：

import static com.mongodb.client.model.Aggregates.*;
import static com.mongodb.client.model.Filters.*;
import static com.mongodb.client.model.Sorts.*;
import static com.mongodb.client.model.Projections.*;

4.1. match + count：统计使用英语的国家数量

需求：统计世界上有多少国家将英语作为官方语言。

@Test
public void givenCountryCollection_whenEnglishSpeakingCountriesCounted_thenNinetyOne() {
    Document result = collection.aggregate(Arrays.asList(
        match(Filters.eq("languages.name", "English")),
        count()
    )).first();

    assertEquals(91, result.get("count"));
}

✅ 关键点解析：

• match 阶段过滤出 languages.name 包含 "English" 的文档。
• count 阶段统计上一阶段的文档数。
• first() 获取单条结果（count 总是返回一个文档）。

⚠️ 踩坑提示：count 返回的是 { "count": 91 } 这样的文档，不是原始数据。

4.2. group + sum + sort：找出国家最多的地理区域

需求：哪个地理区域（如 Africa、Asia）包含的国家最多？

@Test
public void givenCountryCollection_whenCountedRegionWise_thenMaxInAfrica() {
    Document result = collection.aggregate(Arrays.asList(
        group("$region", Accumulators.sum("tally", 1)),
        sort(Sorts.descending("tally"))
    )).first();

    assertTrue(result.getString("_id").equals("Africa"));
}

✅ 关键点解析：

• group("$region", ...) 按 region 字段分组。
• Accumulators.sum("tally", 1) 对每组计数，结果字段名为 tally。
• sort(descending("tally")) 按计数降序排列。
• first() 取第一条即为最大值。

⚠️ 注意：group 的分组字段用 $field 语法，结果中分组值在 _id 字段。

4.3. sort + limit + out：创建“面积最大七国”集合

需求：找出面积最大的7个国家，并存入新集合 largest_seven。

@Test
public void givenCountryCollection_whenAreaSortedDescending_thenSuccess() {
    collection.aggregate(Arrays.asList(
        sort(Sorts.descending("area")),
        limit(7),
        out("largest_seven")
    )).toCollection(); // 执行并写入

    MongoCollection<Document> largestSeven = database.getCollection("largest_seven");
    assertEquals(7, largestSeven.countDocuments());

    Document usa = largestSeven.find(Filters.eq("alpha3Code", "USA")).first();
    assertNotNull(usa);
}

✅ 关键点解析：

• sort(descending("area")) 按面积降序。
• limit(7) 只取前7条。
• out("largest_seven") 将结果写入新集合。
• toCollection() 触发执行（返回 Void）。

✅ out 阶段的威力：结果持久化，可被其他查询或应用直接使用。

4.4. project + group(max) + match：找出邻国最多的国家

需求：计算每个国家的邻国数量，并找出邻国最多的国家。

挑战：数据中 borders 是数组，没有现成的计数字段，需用 project 计算。

@Test
public void givenCountryCollection_whenNeighborsCalculated_thenMaxIsFifteenInChina() {
    // 第一步：投影，计算每个国家的邻国数量
    Bson projectStage = project(fields(
        excludeId(),
        include("name"),
        computed("borderingCountries", computed("$size", "$borders"))
    ));

    // 第二步：分组求最大值（仅返回数值）
    int maxCount = collection.aggregate(Arrays.asList(
        projectStage,
        group(null, Accumulators.max("max", "$borderingCountries"))
    )).first().getInteger("max");

    assertEquals(15, maxCount); // 中国有15个邻国

    // 第三步：匹配出邻国数为15的国家
    Document result = collection.aggregate(Arrays.asList(
        projectStage,
        match(eq("borderingCountries", maxCount))
    )).first();

    assertTrue(result.getString("name").contains("China"));
}

✅ 关键点解析：

• project 使用 $size 操作符计算 borders 数组长度。
• group(null, max(...)) 对整个集合求最大值（null 表示不分组）。
• Accumulators.max() 返回的是最大数值，不是原始文档。
• 因此需再次 match 才能查出是哪个国家。

⚠️ 踩坑提示：聚合结果是“脱水”后的数据，如需原始文档信息，必须在 project 中保留或后续 lookup。

5. 总结

本文通过四个实战示例，演示了如何使用 Java 驱动操作 MongoDB 聚合框架：

• match + count：基础过滤与统计
• group + sum + sort：分组聚合分析
• sort + limit + out：排序取Top N并持久化
• project + group(max) + match：字段计算与反查

✅ 聚合框架的强大之处在于其流水线式设计，可以组合出极其复杂的分析逻辑。

进阶建议：若项目使用 Spring，可考虑 Spring Data MongoDB 提供的聚合支持，代码更简洁。

源码已托管至 GitHub：https://github.com/tech-tutorial/java-mongodb-aggregation