[BEAM-6241] Add support for aggregates using withQueryFn to MongoDbIO

• Type: Improvement
• Status: Resolved
• Priority: P2
• Resolution: Fixed
• Affects Version/s: 2.9.0
• Fix Version/s: 2.12.0
• Component/s: io-java-mongodb
• Labels: easyfix
• Assignee: Ahmed El.Hussaini
• Reporter: Ahmed El.Hussaini
• Created: 16/Dec/18 17:59
• Updated: 23/Apr/21 14:24
• Resolved: 27/Mar/19 20:11

Description

Adds Support to Limit Results

 

MongoDbIO.read()
.withUri("mongodb://localhost:" + port)
.withDatabase(DATABASE)
.withCollection(COLLECTION)
.withFilter("{\"scientist\":\"Einstein\"}")
.withLimit(5));

Adds Support to Use Aggregates

 

List<BsonDocument> aggregates = new ArrayList<BsonDocument>();
  aggregates.add(
    new BsonDocument(
      "$match",
      new BsonDocument("country", new BsonDocument("$eq", new BsonString("England")))));

PCollection<Document> output =
  pipeline.apply(
    MongoDbIO.read()
      .withUri("mongodb://localhost:" + port)
      .withDatabase(DATABASE)
      .withCollection(COLLECTION)
      .withAggregate(aggregates));

 

Issue Links

• is related to: BEAM-12217 MongoDbIO: remove Read.withFilter() and Read.withProjection()
• relates to: BEAM-4567 Can't use mongo connector with Atlas MongoDB
• relates to: BEAM-6521 Deprecate/Remove MongoDb `withFilter` because its contract is broken
• links to: GitHub Pull Request #7293
• links to: https://github.com/apache/beam/pull/7293