在不检索行的情况下执行 MongoTemplate.aggregate答案

【问题标题】：Execute MongoTemplate.aggregate without row retrival在不检索行的情况下执行 MongoTemplate.aggregate
【发布时间】：2020-02-25 19:55:21
【问题描述】：

我正在使用 Spring Mongo 驱动程序执行将运行一段时间的大型 mongo 聚合语句。此聚合的输出阶段将聚合的输出写入新集合。在任何时候我都不需要在内存中检索此聚合的结果。

当我在 Spring boot 中运行此程序时，JVM 在进行行检索时内存不足，尽管我没有使用或存储任何结果。

有没有办法使用 MongoTemplate.aggregate 跳过行检索？

例如：

mongoTemplate.aggregate(Aggregation.newAggregation(
   Aggregation.sort(new Sort(new Sort.Order(Sort.Direction.DESC, "createdOn"))),

   Aggregation.group("accountId")
      .first("bal").as("bal")
      .first("timestamp").as("effectiveTimestamp"),

   Aggregation.project("_id", "effectiveTimestamp")
                        .andExpression("trunc(bal * 10000 + 0.5) / 100").as("bal"),

   aggregationOperationContext -> new Document("$addFields", new Document("history",Arrays.asList(historyObj))),

   // Write results out to a new collection - Do not store in memory
   Aggregation.out("newBalance")
).withOptions(Aggregation.newAggregationOptions().allowDiskUse(true).build()),
   "account", Object.class
);

【问题讨论】：

>有没有办法使用 MongoTemplate.aggregate 跳过行检索？ - 我担心，不是（当你问得那么严格时）。 ..问题可能不在于Aggregation.out，而是Arrays.asList() ...？ 1.historyObj有多大？ 2. 有堆栈跟踪吗？ ...增加内存当然“永远值得”
Arrays.asList(historyObj) 将始终是用于构建此查询的 1 元素数组

标签： java spring mongodb spring-mongodb spring-mongo

【解决方案1】：

使用 AggregationOption - skipOutput() 。如果聚合管道包含 $out/$merge 操作，则不会返回结果。

mongoTemplate.aggregate(aggregation.withOptions(newAggregationOptions().skipOutput().allowDiskUse(true).build()), "collectionNme", EntityClass.class);

如果您使用的是没有框架的 MongoDriver。

MongoClient client = MongoClients.create("mongodb://localhost:27017");
 MongoDatabase database = client.getDatabase("my-collection");
 MongoCollection<Document> model = database.getCollection(collectionName);
 AggregateIterable<Document> aggregateResult = model.aggregate(bsonListOfAggregationPipeline);
 
 // instead iterating over call toCollection() to skipResult
 aggregateIterable.toCollection();

参考资料：

【讨论】：

【解决方案2】：

我可以通过使用来解决这个问题

MongoTempalte.aggregateStream(...).withOptions(Aggregation.newAggregationOptions().cursorBatchSize(0).build)

【讨论】：