releases/2.4.0/storm-mongodb.md - storm-site - Git at Google

 ---
 title: Storm MongoDB Integration
 layout: documentation
 documentation: true
 ---

 Storm/Trident integration for [MongoDB](https://www.mongodb.org/). This package includes the core bolts and trident states that allows a storm topology to either insert storm tuples in a database collection or to execute update queries against a database collection in a storm topology.

 ## Insert into Database
 The bolt and trident state included in this package for inserting data into a database collection.

 ### MongoMapper
 The main API for inserting data in a collection using MongoDB is the `org.apache.storm.mongodb.common.mapper.MongoMapper` interface:

 ```java
 public interface MongoMapper extends Serializable {
     Document toDocument(ITuple tuple);
     Document toDocumentByKeys(List<Object> keys);
 }
 ```

 ### SimpleMongoMapper
 `storm-mongodb` includes a general purpose `MongoMapper` implementation called `SimpleMongoMapper` that can map Storm tuple to a Database document.  `SimpleMongoMapper` assumes that the storm tuple has fields with same name as the document field name in the database collection that you intend to write to.

 ```java
 public class SimpleMongoMapper implements MongoMapper {
     private String[] fields;

     @Override
     public Document toDocument(ITuple tuple) {
         Document document = new Document();
         for(String field : fields){
             document.append(field, tuple.getValueByField(field));
         }
         return document;
     }

     @Override
     public Document toDocumentByKeys(List<Object> keys) {
         Document document = new Document();
         document.append("_id", MongoUtils.getID(keys));
         return document;
     }

     public SimpleMongoMapper withFields(String... fields) {
         this.fields = fields;
         return this;
     }
 }
 ```

 ### MongoInsertBolt
 To use the `MongoInsertBolt`, you construct an instance of it by specifying url, collectionName and a `MongoMapper` implementation that converts storm tuple to DB document. The following is the standard URI connection scheme:
  `mongodb://[username:password@]host1[:port1][,host2[:port2],...[,hostN[:portN]]][/[database][?options]]`

 More options information(eg: Write Concern Options) about Mongo URI, you can visit https://docs.mongodb.org/manual/reference/connection-string/#connections-connection-options

  ```java
 String url = "mongodb://127.0.0.1:27017/test";
 String collectionName = "wordcount";

 MongoMapper mapper = new SimpleMongoMapper()
         .withFields("word", "count");

 MongoInsertBolt insertBolt = new MongoInsertBolt(url, collectionName, mapper);
  ```


 ## Update from Database
 The bolt included in this package for updating data from a database collection.

 ### MongoUpdateMapper
 The main API for updating data in a collection using MongoDB is the `org.apache.storm.mongodb.common.mapper.MongoUpdateMapper` interface:

 ```java
 public interface MongoUpdateMapper extends MongoMapper { }
 ```

 ### SimpleMongoUpdateMapper
 `storm-mongodb` includes a general purpose `MongoUpdateMapper` implementation called `SimpleMongoUpdateMapper` that can map Storm tuple to a Database document. `SimpleMongoUpdateMapper` assumes that the storm tuple has fields with same name as the document field name in the database collection that you intend to write to.
 `SimpleMongoUpdateMapper` uses `$set` operator for setting the value of a field in a document. More information about update operator, you can visit
 https://docs.mongodb.org/manual/reference/operator/update/

 ```java
 public class SimpleMongoUpdateMapper extends SimpleMongoMapper implements MongoUpdateMapper {

     private String[] fields;

     @Override
     public Document toDocument(ITuple tuple) {
         Document document = new Document();
         for(String field : fields){
             document.append(field, tuple.getValueByField(field));
         }
         //$set operator: Sets the value of a field in a document.
         return new Document("$set", document);
     }

     public SimpleMongoUpdateMapper withFields(String... fields) {
         this.fields = fields;
         return this;
     }
 }
 ```


 ### QueryFilterCreator
 The main API for creating a MongoDB query Filter is the `org.apache.storm.mongodb.common.QueryFilterCreator` interface:

  ```java
 public interface QueryFilterCreator extends Serializable {
     Bson createFilter(ITuple tuple);
     Bson createFilterByKeys(List<Object> keys);
 }
  ```

 ### SimpleQueryFilterCreator
 `storm-mongodb` includes a general purpose `QueryFilterCreator` implementation called `SimpleQueryFilterCreator` that can create a MongoDB query Filter by given Tuple.  `QueryFilterCreator` uses `$eq` operator for matching values that are equal to a specified value. More information about query operator, you can visit
 https://docs.mongodb.org/manual/reference/operator/query/

  ```java
 public class SimpleQueryFilterCreator implements QueryFilterCreator {

     private String field;

     @Override
     public Bson createFilter(ITuple tuple) {
         return Filters.eq(field, tuple.getValueByField(field));
     }

     @Override
     public Bson createFilterByKeys(List<Object> keys) {
         return Filters.eq("_id", MongoUtils.getID(keys));
     }

     public SimpleQueryFilterCreator withField(String field) {
         this.field = field;
         return this;
     }

 }
  ```

 ### MongoUpdateBolt
 To use the `MongoUpdateBolt`,  you construct an instance of it by specifying Mongo url, collectionName, a `QueryFilterCreator` implementation and a `MongoUpdateMapper` implementation that converts storm tuple to DB document.

  ```java
         MongoUpdateMapper mapper = new SimpleMongoUpdateMapper()
                 .withFields("word", "count");

         QueryFilterCreator updateQueryCreator = new SimpleQueryFilterCreator()
                 .withField("word");

         MongoUpdateBolt updateBolt = new MongoUpdateBolt(url, collectionName, updateQueryCreator, mapper);

         //if a new document should be inserted if there are no matches to the query filter
         //updateBolt.withUpsert(true);

         //whether find all documents according to the query filter
         //updateBolt.withMany(true);
  ```

  Or use a anonymous inner class implementation for `QueryFilterCreator`:

   ```java
         MongoUpdateMapper mapper = new SimpleMongoUpdateMapper()
                 .withFields("word", "count");

         QueryFilterCreator updateQueryCreator = new QueryFilterCreator() {
             @Override
             public Bson createFilter(ITuple tuple) {
                 return Filters.gt("count", 3);
             }
         };

         MongoUpdateBolt updateBolt = new MongoUpdateBolt(url, collectionName, updateQueryCreator, mapper);

         //if a new document should be inserted if there are no matches to the query filter
         //updateBolt.withUpsert(true);
  ```


 ## Lookup from Database
 The bolt included in this package for selecting data from a database collection.

 ### MongoLookupMapper
 The main API for selecting data in a collection using MongoDB is the `org.apache.storm.mongodb.common.mapper.MongoLookupMapper` interface:

 ```java
 public interface MongoLookupMapper extends Serializable {

     List<Values> toTuple(ITuple input, Document doc);

     void declareOutputFields(OutputFieldsDeclarer declarer);
 }
 ```

 ### SimpleMongoLookupMapper
 `storm-mongodb` includes a general purpose `MongoLookupMapper` implementation called `SimpleMongoLookupMapper` that can converts a Mongo document to a list of storm values.

 ```java
 public class SimpleMongoLookupMapper implements MongoLookupMapper {

     private String[] fields;

     @Override
     public List<Values> toTuple(ITuple input, Document doc) {
         Values values = new Values();

         for(String field : fields) {
             if(input.contains(field)) {
                 values.add(input.getValueByField(field));
             } else {
                 values.add(doc.get(field));
             }
         }
         List<Values> result = new ArrayList<Values>();
         result.add(values);
         return result;
     }

     @Override
     public void declareOutputFields(OutputFieldsDeclarer declarer) {
         declarer.declare(new Fields(fields));
     }

     public SimpleMongoLookupMapper withFields(String... fields) {
         this.fields = fields;
         return this;
     }

 }
 ```

 ### MongoLookupBolt
 To use the `MongoLookupBolt`,  you construct an instance of it by specifying Mongo url, collectionName, a `QueryFilterCreator` implementation and a `MongoLookupMapper` implementation that converts a Mongo document to a list of storm values.

  ```java
         MongoLookupMapper mapper = new SimpleMongoLookupMapper()
                 .withFields("word", "count");

         QueryFilterCreator filterCreator = new SimpleQueryFilterCreator()
                 .withField("word");

         MongoLookupBolt lookupBolt = new MongoLookupBolt(url, collectionName, filterCreator, mapper);
  ```

 ## Mongo Trident State&MapState
 ### Trident State
 We support trident persistent state that can be used with trident topologies. To create a Mongo persistent trident state you need to initialize it with the url, collectionName, the `MongoMapper` instance. See the example below:

  ```java
         MongoMapper mapper = new SimpleMongoMapper()
                 .withFields("word", "count");

         MongoState.Options options = new MongoState.Options()
                 .withUrl(url)
                 .withCollectionName(collectionName)
                 .withMapper(mapper);

         StateFactory factory = new MongoStateFactory(options);

         TridentTopology topology = new TridentTopology();
         Stream stream = topology.newStream("spout1", spout);

         stream.partitionPersist(factory, fields,
                 new MongoStateUpdater(), new Fields());

         TridentState state = topology.newStaticState(factory);
         stream = stream.stateQuery(state, new Fields("word"),
                 new MongoStateQuery(), new Fields("columnName", "columnValue"));
         stream.each(new Fields("word", "columnValue"), new PrintFunction(), new Fields());
  ```
  **NOTE**:
  >If there is no unique index provided, trident state inserts in the case of failures may result in duplicate documents.

 ### Trident MapState
 We also support trident `MapState`. To create a Mongo trident `MapState` you need to initialize it with the url, collectionName, the `MongoMapper` and `QueryFilterCreator` instance. See the example below:

  ```java
         MongoMapper mapper = new SimpleMongoMapper()
                 .withFields("word", "count");

         QueryFilterCreator filterCreator = new SimpleQueryFilterCreator()
                 .withField("word");

         MongoMapState.Options options = new MongoMapState.Options();
         options.url = url;
         options.collectionName = collectionName;
         options.mapper = mapper;
         options.queryCreator = filterCreator;

         StateFactory factory = MongoMapState.transactional(options);

         TridentTopology topology = new TridentTopology();
         Stream stream = topology.newStream("spout1", spout);

         TridentState state = stream.groupBy(new Fields("word"))
                 .persistentAggregate(factory, new Fields("count"), new Sum(), new Fields("sum"));

         stream.stateQuery(state, new Fields("word"), new MapGet(), new Fields("sum"))
                 .each(new Fields("word", "sum"), new PrintFunction(), new Fields());
  ```
	---
	title: Storm MongoDB Integration
	layout: documentation
	documentation: true
	---

	Storm/Trident integration for [MongoDB](https://www.mongodb.org/). This package includes the core bolts and trident states that allows a storm topology to either insert storm tuples in a database collection or to execute update queries against a database collection in a storm topology.

	## Insert into Database
	The bolt and trident state included in this package for inserting data into a database collection.

	### MongoMapper
	The main API for inserting data in a collection using MongoDB is the `org.apache.storm.mongodb.common.mapper.MongoMapper` interface:

	```java
	public interface MongoMapper extends Serializable {
	Document toDocument(ITuple tuple);
	Document toDocumentByKeys(List<Object> keys);
	}
	```

	### SimpleMongoMapper
	`storm-mongodb` includes a general purpose `MongoMapper` implementation called `SimpleMongoMapper` that can map Storm tuple to a Database document. `SimpleMongoMapper` assumes that the storm tuple has fields with same name as the document field name in the database collection that you intend to write to.

	```java
	public class SimpleMongoMapper implements MongoMapper {
	private String[] fields;

	@Override
	public Document toDocument(ITuple tuple) {
	Document document = new Document();
	for(String field : fields){
	document.append(field, tuple.getValueByField(field));
	}
	return document;
	}

	@Override
	public Document toDocumentByKeys(List<Object> keys) {
	Document document = new Document();
	document.append("_id", MongoUtils.getID(keys));
	return document;
	}

	public SimpleMongoMapper withFields(String... fields) {
	this.fields = fields;
	return this;
	}
	}
	```

	### MongoInsertBolt
	To use the `MongoInsertBolt`, you construct an instance of it by specifying url, collectionName and a `MongoMapper` implementation that converts storm tuple to DB document. The following is the standard URI connection scheme:
	`mongodb://[username:password@]host1[:port1][,host2[:port2],...[,hostN[:portN]]][/[database][?options]]`

	More options information(eg: Write Concern Options) about Mongo URI, you can visit https://docs.mongodb.org/manual/reference/connection-string/#connections-connection-options

	```java
	String url = "mongodb://127.0.0.1:27017/test";
	String collectionName = "wordcount";

	MongoMapper mapper = new SimpleMongoMapper()
	.withFields("word", "count");

	MongoInsertBolt insertBolt = new MongoInsertBolt(url, collectionName, mapper);
	```


	## Update from Database
	The bolt included in this package for updating data from a database collection.

	### MongoUpdateMapper
	The main API for updating data in a collection using MongoDB is the `org.apache.storm.mongodb.common.mapper.MongoUpdateMapper` interface:

	```java
	public interface MongoUpdateMapper extends MongoMapper { }
	```

	### SimpleMongoUpdateMapper
	`storm-mongodb` includes a general purpose `MongoUpdateMapper` implementation called `SimpleMongoUpdateMapper` that can map Storm tuple to a Database document. `SimpleMongoUpdateMapper` assumes that the storm tuple has fields with same name as the document field name in the database collection that you intend to write to.
	`SimpleMongoUpdateMapper` uses `$set` operator for setting the value of a field in a document. More information about update operator, you can visit
	https://docs.mongodb.org/manual/reference/operator/update/

	```java
	public class SimpleMongoUpdateMapper extends SimpleMongoMapper implements MongoUpdateMapper {

	private String[] fields;

	@Override
	public Document toDocument(ITuple tuple) {
	Document document = new Document();
	for(String field : fields){
	document.append(field, tuple.getValueByField(field));
	}
	//$set operator: Sets the value of a field in a document.
	return new Document("$set", document);
	}

	public SimpleMongoUpdateMapper withFields(String... fields) {
	this.fields = fields;
	return this;
	}
	}
	```


	### QueryFilterCreator
	The main API for creating a MongoDB query Filter is the `org.apache.storm.mongodb.common.QueryFilterCreator` interface:

	```java
	public interface QueryFilterCreator extends Serializable {
	Bson createFilter(ITuple tuple);
	Bson createFilterByKeys(List<Object> keys);
	}
	```

	### SimpleQueryFilterCreator
	`storm-mongodb` includes a general purpose `QueryFilterCreator` implementation called `SimpleQueryFilterCreator` that can create a MongoDB query Filter by given Tuple. `QueryFilterCreator` uses `$eq` operator for matching values that are equal to a specified value. More information about query operator, you can visit
	https://docs.mongodb.org/manual/reference/operator/query/

	```java
	public class SimpleQueryFilterCreator implements QueryFilterCreator {

	private String field;

	@Override
	public Bson createFilter(ITuple tuple) {
	return Filters.eq(field, tuple.getValueByField(field));
	}

	@Override
	public Bson createFilterByKeys(List<Object> keys) {
	return Filters.eq("_id", MongoUtils.getID(keys));
	}

	public SimpleQueryFilterCreator withField(String field) {
	this.field = field;
	return this;
	}

	}
	```

	### MongoUpdateBolt
	To use the `MongoUpdateBolt`, you construct an instance of it by specifying Mongo url, collectionName, a `QueryFilterCreator` implementation and a `MongoUpdateMapper` implementation that converts storm tuple to DB document.

	```java
	MongoUpdateMapper mapper = new SimpleMongoUpdateMapper()
	.withFields("word", "count");

	QueryFilterCreator updateQueryCreator = new SimpleQueryFilterCreator()
	.withField("word");

	MongoUpdateBolt updateBolt = new MongoUpdateBolt(url, collectionName, updateQueryCreator, mapper);

	//if a new document should be inserted if there are no matches to the query filter
	//updateBolt.withUpsert(true);

	//whether find all documents according to the query filter
	//updateBolt.withMany(true);
	```

	Or use a anonymous inner class implementation for `QueryFilterCreator`:

	```java
	MongoUpdateMapper mapper = new SimpleMongoUpdateMapper()
	.withFields("word", "count");

	QueryFilterCreator updateQueryCreator = new QueryFilterCreator() {
	@Override
	public Bson createFilter(ITuple tuple) {
	return Filters.gt("count", 3);
	}
	};

	MongoUpdateBolt updateBolt = new MongoUpdateBolt(url, collectionName, updateQueryCreator, mapper);

	//if a new document should be inserted if there are no matches to the query filter
	//updateBolt.withUpsert(true);
	```


	## Lookup from Database
	The bolt included in this package for selecting data from a database collection.

	### MongoLookupMapper
	The main API for selecting data in a collection using MongoDB is the `org.apache.storm.mongodb.common.mapper.MongoLookupMapper` interface:

	```java
	public interface MongoLookupMapper extends Serializable {

	List<Values> toTuple(ITuple input, Document doc);

	void declareOutputFields(OutputFieldsDeclarer declarer);
	}
	```

	### SimpleMongoLookupMapper
	`storm-mongodb` includes a general purpose `MongoLookupMapper` implementation called `SimpleMongoLookupMapper` that can converts a Mongo document to a list of storm values.

	```java
	public class SimpleMongoLookupMapper implements MongoLookupMapper {

	private String[] fields;

	@Override
	public List<Values> toTuple(ITuple input, Document doc) {
	Values values = new Values();

	for(String field : fields) {
	if(input.contains(field)) {
	values.add(input.getValueByField(field));
	} else {
	values.add(doc.get(field));
	}
	}
	List<Values> result = new ArrayList<Values>();
	result.add(values);
	return result;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
	declarer.declare(new Fields(fields));
	}

	public SimpleMongoLookupMapper withFields(String... fields) {
	this.fields = fields;
	return this;
	}

	}
	```

	### MongoLookupBolt
	To use the `MongoLookupBolt`, you construct an instance of it by specifying Mongo url, collectionName, a `QueryFilterCreator` implementation and a `MongoLookupMapper` implementation that converts a Mongo document to a list of storm values.

	```java
	MongoLookupMapper mapper = new SimpleMongoLookupMapper()
	.withFields("word", "count");

	QueryFilterCreator filterCreator = new SimpleQueryFilterCreator()
	.withField("word");

	MongoLookupBolt lookupBolt = new MongoLookupBolt(url, collectionName, filterCreator, mapper);
	```

	## Mongo Trident State&MapState
	### Trident State
	We support trident persistent state that can be used with trident topologies. To create a Mongo persistent trident state you need to initialize it with the url, collectionName, the `MongoMapper` instance. See the example below:

	```java
	MongoMapper mapper = new SimpleMongoMapper()
	.withFields("word", "count");

	MongoState.Options options = new MongoState.Options()
	.withUrl(url)
	.withCollectionName(collectionName)
	.withMapper(mapper);

	StateFactory factory = new MongoStateFactory(options);

	TridentTopology topology = new TridentTopology();
	Stream stream = topology.newStream("spout1", spout);

	stream.partitionPersist(factory, fields,
	new MongoStateUpdater(), new Fields());

	TridentState state = topology.newStaticState(factory);
	stream = stream.stateQuery(state, new Fields("word"),
	new MongoStateQuery(), new Fields("columnName", "columnValue"));
	stream.each(new Fields("word", "columnValue"), new PrintFunction(), new Fields());
	```
	NOTE:
	>If there is no unique index provided, trident state inserts in the case of failures may result in duplicate documents.

	### Trident MapState
	We also support trident `MapState`. To create a Mongo trident `MapState` you need to initialize it with the url, collectionName, the `MongoMapper` and `QueryFilterCreator` instance. See the example below:

	```java
	MongoMapper mapper = new SimpleMongoMapper()
	.withFields("word", "count");

	QueryFilterCreator filterCreator = new SimpleQueryFilterCreator()
	.withField("word");

	MongoMapState.Options options = new MongoMapState.Options();
	options.url = url;
	options.collectionName = collectionName;
	options.mapper = mapper;
	options.queryCreator = filterCreator;

	StateFactory factory = MongoMapState.transactional(options);

	TridentTopology topology = new TridentTopology();
	Stream stream = topology.newStream("spout1", spout);

	TridentState state = stream.groupBy(new Fields("word"))
	.persistentAggregate(factory, new Fields("count"), new Sum(), new Fields("sum"));

	stream.stateQuery(state, new Fields("word"), new MapGet(), new Fields("sum"))
	.each(new Fields("word", "sum"), new PrintFunction(), new Fields());
	```