Apache beam 中 MongoDB 聚合查询的使用

Question

我正在尝试使用聚合查询作为 Apache Beam MongoDBIO QueryFn 的一部分。但我没有得到任何结果。

List<BsonDocument> documents = new ArrayList<>();
documents.add(
       new BsonDocument(
              "$group",
               new BsonDocument("_id", new BsonString("$field_name1"))
                   .append("count", new BsonDocument("$sum", new BsonString("field_name2")))
         ));

 pipeline.apply(MongoDbIO.read()
          .withUri("mongodb://localhost:27017")
          .withDatabase("databaseName")
          .withCollection("collectionName")
          .withQueryFn(AggregationQuery.create().withMongoDbPipeline(documents)));

查询结构是否正确？

Answer 1

您应该为管道中的每个字段使用适当的 BsonValue 子类，而不是使用 BsonString。例如，要指定一个字符串值，您应该使用 BsonString。要指定数值，您应该使用 BsonInt32、BsonInt64 或其他数值类型，具体取决于您的字段的数据类型。

List<BsonDocument> documents = new ArrayList<>();
documents.add(
    new BsonDocument(
        "$group",
        new BsonDocument("_id", new BsonString("$field_name1"))
            .append("count", new BsonDocument("$sum", new BsonInt32(1))) // use BsonInt32 instead of BsonString
    )
);

pipeline.apply(MongoDbIO.read()
    .withUri("mongodb://localhost:27017")
    .withDatabase("databaseName")
    .withCollection("collectionName")
    .withQueryFn(AggregationQuery.create().withMongoDbPipeline(documents))
);