org.apache.beam.sdk.io.mongodb

Class MongoDbIO

java.lang.Object

org.apache.beam.sdk.io.mongodb.MongoDbIO

public class MongoDbIO
extends java.lang.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") )));
 *
 

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	MongoDbIO.Read A PTransform to read data from MongoDB.
static class	MongoDbIO.Write A PTransform to write to a MongoDB database.

Method Summary

Modifier and Type	Method and Description
static MongoDbIO.Read	read() Read data from MongoDB.
static MongoDbIO.Write	write() Write data to MongoDB.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Detail

read

public static MongoDbIO.Read read()

Read data from MongoDB.

write

public static MongoDbIO.Write write()

Write data to MongoDB.