Package org.apache.beam.sdk.io.mongodb

Class MongoDbIO

java.lang.Object
org.apache.beam.sdk.io.mongodb.MongoDbIO
public class MongoDbIO extends Object
IO to read and write data on MongoDB.

Reading from MongoDB

MongoDbIO source returns a bounded collection of String as PCollection<String>. The String is the JSON form of the MongoDB Document.

To configure the MongoDB source, you have to provide the connection URI, the database name and the collection name. The following example illustrates various options for configuring the source: 


 pipeline.apply(MongoDbIO.read()
   .withUri("mongodb://localhost:27017")
   .withDatabase("my-database")
   .withCollection("my-collection"))
   // above three are required configuration, returns PCollection<String>

   // rest of the settings are optional

The source also accepts an optional configuration: withFilter() allows you to define a JSON filter to get subset of data.

Writing to MongoDB

MongoDB sink supports writing of Document (as JSON String) in a MongoDB.

To configure a MongoDB sink and insert/replace, you must specify a connection URI, a Database name, a Collection name. For instance: 


 pipeline
   .apply(...)
   .apply(MongoDbIO.write()
     .withUri("mongodb://localhost:27017")
     .withDatabase("my-database")
     .withCollection("my-collection")
     .withNumSplits(30))
*

To configure a MongoDB sink and update, you must specify a connection URI, a Database * name, a Collection name. It matches the key with _id in target collection. For instance: * * 


 * pipeline
 *   .apply(...)
 *   .apply(MongoDbIO.write()
 *     .withUri("mongodb://localhost:27017")
 *     .withDatabase("my-database")
 *     .withCollection("my-collection")
 *     .withUpdateConfiguration(UpdateConfiguration.create().withFindKey("key1").withUpdateKey("key2")
 *     .withUpdateFields(UpdateField.fieldUpdate("$set", "source-field1", "dest-field1"),
 *                       UpdateField.fieldUpdate("$set","source-field2", "dest-field2"),
 *                       //pushes entire input doc to the dest field
 *                        UpdateField.fullUpdate("$push", "dest-field3") )));
 *