MongoDB聚合：获取重复数据完整集方法

本文针对MongoDB聚合查询中数据去重问题，提供解决方案。通常，由于`$group`阶段的默认行为，聚合管道会移除重复数据，导致结果集不完整。本文分析了`$group`操作的机制，并通过移除该阶段或采用更合适的查询方法（如`find`方法），演示了如何确保在获取匹配文档时完整保留所有重复数据。此外，文章还提供了代码示例和最佳实践，旨在帮助开发者准确获取所需的包含重复数据的完整结果集，同时避免不必要的性能损耗，从而优化MongoDB查询效率。

本文旨在解决MongoDB聚合查询中因不当使用group阶段导致数据去重的问题。通过分析group操作的机制，我们揭示了其如何默认移除重复项。教程将演示如何通过移除group阶段或采用更合适的查询方法，确保在获取匹配文档时完整保留所有重复数据，并提供代码示例与最佳实践，帮助开发者准确获取所需结果。

1. MongoDB聚合管道与数据去重机制

MongoDB的聚合管道（Aggregation Pipeline）是一个强大的数据处理框架，允许用户通过一系列阶段（Stages）对文档进行转换和处理。其中，$group 阶段是实现数据聚合和统计的核心。它的主要作用是根据一个或多个指定的字段（_id 表达式）对文档进行分组，然后对每个组应用累加器（Accumulator）表达式来计算聚合结果。

需要注意的是，$group 阶段的默认行为是为每个唯一的 _id 表达式生成一个输出文档。这意味着，如果我们将 _id 设置为某个字段（例如 numBerId），那么最终结果中，该字段的每个唯一值将只对应一个文档，从而有效地移除了该字段上的“重复”数据。这在需要获取唯一值或进行统计分析时非常有用，但在需要保留所有匹配文档（包括那些具有相同字段值的文档）时，则会导致数据丢失。

2. 原始查询问题分析：为何丢失重复数据

在开发实践中，有时我们需要从MongoDB中检索所有符合条件的文档，包括那些在某个特定字段上值相同的文档。然而，如果不了解 $group 阶段的工作原理，可能会不慎引入它，导致结果集中的数据被去重。

考虑以下Java代码示例，它使用Spring Data MongoDB的 TypedAggregation 来构建一个聚合管道：

Aggregation agg = TypedAggregation.newAggregation(
        TypedAggregation.match(Criteria.where("numBerId").regex("^" + numBerId, "i")
                .andOperator(Criteria.where("numBerId").ne(""))),
        TypedAggregation.group("numBerId"), // 这一步导致了去重
        TypedAggregation.limit(20000),
        TypedAggregation.sort(Direction.ASC, "_id"));

Document rawResults = mongo.aggregate(agg, collectionName(), Document.class).getRawResults();
return rawResults.getList("results", Document.class)
        .stream()
        .map(d -> (String) d.get("_id"))
        .collect(Collectors.toList());

上述代码的目标是根据 numBerId 字段进行模糊匹配，并获取所有匹配的 numBerId。然而，问题出在 TypedAggregation.group("numBerId") 这一行。这个 group 阶段会以 numBerId 字段作为分组键。这意味着，如果多个文档具有相同的 numBerId 值，它们将被分到同一个组，并且 group 阶段只会为这个唯一的 numBerId 输出一个结果文档。最终，通过 _id 获取的 numBerId 列表将不包含任何重复项，即使原始集合中存在多个文档具有相同的 numBerId。

3. 解决方案：确保获取包含重复数据的完整结果集

要获取所有匹配的文档，包括那些具有重复 numBerId 值的文档，关键在于避免使用会去重的聚合阶段。

3.1 方法一：移除 group 阶段

最直接的解决方案是简单地从聚合管道中移除 TypedAggregation.group("numBerId") 阶段。如果我们的目标只是过滤文档并可能进行排序或限制数量，那么 group 阶段是不必要的。

修改后的聚合管道代码如下：

import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.domain.Sort.Direction;
import org.bson.Document;
import java.util.List;
import java.util.stream.Collectors;

// 假设 mongo 是 MongoTemplate 实例，collectionName() 返回集合名
public List<String> getAllNumBerIdsWithDuplicates(String numBerIdPattern) {
    Aggregation agg = TypedAggregation.newAggregation(
            TypedAggregation.match(Criteria.where("numBerId").regex("^" + numBerIdPattern, "i")
                    .andOperator(Criteria.where("numBerId").ne(""))),
            // 移除 group 阶段，不再进行去重
            TypedAggregation.limit(20000),
            TypedAggregation.sort(Direction.ASC, "numBerId"), // 排序字段可以调整为 numBerId
            // 如果只需要 numBerId 字段，可以添加一个 project 阶段
            TypedAggregation.project("numBerId")
    );

    // 执行聚合查询
    List<Document> results = mongo.aggregate(agg, collectionName(), Document.class).getMappedResults();

    // 从结果中提取 numBerId 字段
    return results.stream()
            .map(d -> (String) d.get("numBerId")) // 注意这里是获取 "numBerId"，而不是 "_id"
            .collect(Collectors.toList());
}

解释： 通过移除 TypedAggregation.group("numBerId")，聚合管道在 match 阶段过滤出所有符合条件的文档后，会直接将这些文档传递给 limit 和 sort 阶段。如果添加了 project("numBerId")，则每个文档只保留 numBerId 字段。最终，getMappedResults() 返回的文档列表中将包含所有匹配的 numBerId，包括重复项。需要注意的是，在移除 group 后，返回的文档不再有由 group 阶段生成的 _id 字段，而是原始文档的字段，因此在映射时应使用 d.get("numBerId")。

3.2 方法二：使用 find 方法进行简单查询

对于仅仅需要过滤和投影（即选择特定字段）的场景，使用MongoDB的 find 方法通常比聚合管道更简洁、高效。Spring Data MongoDB提供了 MongoTemplate 的 find 方法来支持此类操作。

import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Field;
import org.springframework.data.domain.Sort;
import java.util.List;

public List<String> findNumBerIdsWithDuplicates(String numBerIdPattern) {
    Query query = new Query(Criteria.where("numBerId").regex("^" + numBerIdPattern, "i")
            .andOperator(Criteria.where("numBerId").ne("")));

    // 限制返回字段，只包含 numBerId
    query.fields().include("numBerId");

    // 限制结果数量
    query.limit(20000);

    // 排序
    query.with(Sort.by(Sort.Direction.ASC, "numBerId"));

    // 执行查询，返回只包含 numBerId 字段的文档列表
    // 假设 NumBerIdProjection 是一个包含 String numBerId 字段的POJO
    // 或者直接返回 Document，然后手动映射
    List<Document> results = mongo.find(query, Document.class, collectionName());

    return results.stream()
            .map(d -> (String) d.get("numBerId"))
            .collect(Collectors.toList());
}

解释：Query 对象允许我们构建复杂的查询条件 (Criteria)、指定返回字段 (fields())、设置限制 (limit()) 和排序 (with(Sort.by()))。这种方法在功能上等同于仅包含 match、project、limit 和 sort 的聚合管道，并且通常更易于理解和维护。

4. group 阶段的正确应用场景

虽然 group 阶段会导致数据去重，但它在以下场景中是不可或缺的：

• 统计每个唯一值的出现次数：

  TypedAggregation.newAggregation(
      TypedAggregation.group("numBerId").count().as("count")
  )
  // 结果: [{ "_id": "A123", "count": 5 }, { "_id": "B456", "count": 2 }]

• 计算每个组的总和、平均值、最大值、最小值等：

  TypedAggregation.newAggregation(
      TypedAggregation.group("category")
          .sum("price").as("totalPrice")
          .avg("quantity").as("avgQuantity")
  )

• 获取每个组的第一个/最后一个文档：

  TypedAggregation.newAggregation(
      TypedAggregation.sort(Direction.ASC, "timestamp"), // 确保排序以便获取正确的第一个/最后一个
      TypedAggregation.group("userId")
          .first("event").as("firstEvent")
  )

• 收集每个组的所有值到一个数组中：

  TypedAggregation.newAggregation(
      TypedAggregation.group("category")
          .push("productName").as("products")
  )
  // 结果: [{ "_id": "Electronics", "products": ["TV", "Phone"] }]

当你的目标是基于某个或某些字段进行聚合计算或生成每个唯一分组的汇总信息时，group 阶段才是正确的选择。

5. 注意事项与性能优化

在执行MongoDB查询时，无论使用聚合还是 find 方法，都应考虑以下最佳实践以优化性能和确保数据准确性：

• 索引优化： 在 match 阶段使用的查询字段（如 numBerId）上创建索引是至关重要的。索引可以显著加快查询速度，尤其是在处理大量数据时。

  db.yourCollection.createIndex({ "numBerId": 1 });

• 数据投影： 如果你只需要文档中的部分字段，应始终使用 project 阶段（聚合管道）或 fields().include()（find 方法）来限制返回的字段。这可以减少网络传输的数据量和客户端的内存消耗。
• 内存限制： MongoDB聚合管道默认有100MB的内存限制。对于大型聚合操作，如果可能超过此限制，可以使用 allowDiskUse(true) 选项，让MongoDB将数据写入临时文件以完成操作。但在Spring Data MongoDB中，这通常通过 AggregationOptions 配置。
• 结果集处理： 根据查询的预期结果大小，选择合适的返回类型。对于大型结果集，考虑使用游标（cursor）进行分批处理，而不是一次性加载所有数据到内存。
• 错误处理： 在实际应用中，应加入适当的错误处理机制，例如对查询结果进行空值检查，以及捕获和处理MongoDB操作可能抛出的异常。

通过理解 group 阶段的机制并选择适合需求的查询方法，开发者可以更精确、高效地从MongoDB中获取所需的数据，无论是去重后的统计结果还是包含所有重复项的完整数据集。

好了，本文到此结束，带大家了解了《MongoDB聚合：获取重复数据完整集方法》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多文章知识！