mongodb - Mongo Hadoop 连接器支持聚合吗？

Question

我正在尝试使用 spark 的 mongo hadoop ( https://github.com/mongodb/mongo-hadoop) 库对 mongo 集合执行一些聚合操作。我使用作为输入发送到 newApiHadoopRDD 的 mongo.input.query 配置输入我的查询。

Configuration mongodbConfig = new Configuration();
mongodbConfig.set("mongo.job.input.format","com.mongodb.hadoop.MongoInputFormat";
mongodbConfig.set("mongo.input.uri","mongodb://"+mongodbHost+"/"+database.collection);
mongodbConfig.set("mongo.input.query",query);

JavaPairRDD audienceRDD =  sc.newAPIHadoopRDD(mongodbConfig, MongoInputFormat.class,Object.class, BSONObject.class);
audienceRDD.foreach(e -> System.out.println("data: "+e.toString()));


query={ "aggregate" : "__collection__" , "pipeline" : [ 
{ "$match" : { "date" : { "$gte" : { "$date" : "2016-08-09T00:00:00.000Z"} , "$lte" : { "$date" : "2016-08-11T00:00:00.000Z"}}}} , 
{ "$unwind" : "$segments"} , 
{ "$group" : { "_id" : "$segments" , "audienceSize" : { "$sum" : "$count"}}}]}, sort={ }, fields={ }, limit=0, notimeout=false}

如果我使用像查找这样的普通查询，则操作成功。但是当我尝试使用 groupBy 时，我无法获得 RDD 上的任何记录。有人可以建议一种使用 mongo hadoop 连接器对 mongo 集合进行聚合操作的方法。

Answer 1

无论如何，由于运行聚合查询的 16 MB 限制，我最终创建了一个包含记录的临时集合，然后对该临时集合进行查询。将响应存储在 RDD 上，一旦我完成了我想做的事情，我就删除了临时集合。

也就是说，我认为添加使用 mongo.input.query 进行聚合查询的能力将是对这个不错的连接器库的一个很好的补充。