使用spring mongodb对聚合操作的结果进行流处理。

Question

我正在使用，我想使用一个游标来进行聚合操作。

MongoTemplate.stream()获得一个查询，因此我尝试创建聚合实例，使用Aggregation.toDbObject()将其转换为DbObject，使用DbObject创建一个BasicQuery，然后调用stream()方法。

这将返回一个空游标。

调试streaming代码表明，MongoTemplate.stream()使用FindOperation，这让我想到了spring mongodb不支持流聚合操作。

是否有人能够使用来流聚合查询的结果？

为了记录在案，我可以使用驱动程序来完成这个任务，但我更喜欢使用spring数据。

编辑11月10日-添加示例代码：

    MatchOperation match = Aggregation.match(Criteria.where("type").ne("AType"));
    GroupOperation group = Aggregation.group("name", "type");
    group = group.push("color").as("colors");
    group = group.push("size").as("sizes");
    TypedAggregation<MyClass> agg = Aggregation.newAggregation(MyClass.class, Arrays.asList(match, group));

    MongoConverter converter = mongoTemplate.getConverter();
    MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> mappingContext = converter.getMappingContext();
    QueryMapper queryMapper = new QueryMapper(converter);
    AggregationOperationContext context = new TypeBasedAggregationOperationContext(MyClass.class, mappingContext, queryMapper);
    // create a BasicQuery to be used in the stream() method by converting the Aggregation to a DbObject
    BasicQuery query = new BasicQuery(agg.toDbObject("myClass", context));

    // spring-mongo attributes the stream() method to find() operationsm not to aggregate() operations so the stream returns an empty cursor
    CloseableIterator<MyClass> iter = mongoTemplate.stream(query, MyClass.class);

    // this is an empty cursor
    while(iter.hasNext()) {
        System.out.println(iter.next().getName());
    }

以下代码不使用stream()方法，返回聚合的预期非空结果：

    AggregationResults<HashMap> result = mongoTemplate.aggregate(agg, "myClass", HashMap.class);

Answer 1

对于那些仍在试图找到答案的人来说：

从SpringDataMongoVersion2.0.0.M4开始(AFAIK)，MongoTemplate得到了一个aggregateStream方法。

因此，您可以进行以下操作：

 AggregationOptions aggregationOptions = Aggregation.newAggregationOptions()
        // this is very important: if you do not set the batch size, you'll get all the objects at once and you might run out of memory if the returning data set is too large
        .cursorBatchSize(mongoCursorBatchSize)
        .build();

    data = mongoTemplate.aggregateStream(Aggregation.newAggregation(
            Aggregation.group("person_id").count().as("count")).withOptions(aggregationOptions), collectionName, YourClazz.class);