mongo spark连接器的聚合问题

soat7uwm  于 2021-05-18  发布在  Spark
关注(0)|答案(0)|浏览(206)

我们在使用mongospark连接器的聚合查询时面临一个问题。
我们在spark作业中从mongodb(通过mongosparkapi)读取数据时,在某些字段上使用groupby,但我们发现mongospatk connector在执行groupby查询之前正在对数据进行分区。这会导致不正确的结果。
例如,在下面的查询中,我们期望聚合查询在集合中的完整数据上运行(基于customer\ id字段),然后进行分区。但是,实际输出显示groupby查询在已经分区的数据上运行。

Map<String, String> readOverrides = new HashMap<String, String>();

        readOverrides.put("database", "mydb");
        readOverrides.put("collection", "myCollection");
        readOverrides.put("readPreference.name", "secondaryPreferred");
Document groupFields = new Document("customer_id", "$customer_id");
        JavaMongoRDD<Document> myRDD = MongoSpark.load(jsc, readConfig);
        Dataset<Row> dsr = myRDD
                .withPipeline(Arrays.asList(new Document("$match", new Document("date", date)),
                        new Document("$group", new Document("_id", "$customer_id").append("field1", new Document("$first","$field1")).
                                .append("sumField2", new Document("$sum","$field2"))
                        , new Document("$skip",0)
                        , new Document("$limit", 1000)
                        )
                ).toDF();
            dsr1.show();

我们正在使用这些版本-spark v2.4.6 mongospark connector v2.4.2。寻找任何关于这个的建议。谢谢

暂无答案!

目前还没有任何答案,快来回答吧!

相关问题