MongoDB

This page details how to set up MongoDB change stream.

1. Overview
The MongoDB change stream works similar to Cinchy's Change Data Capture functionality. The listener subscribes to monitor the change stream of a specific collection in the database of the MongoDB server. Any actions performed on document(s) inside of that collection are picked up by the listener and sent to the queue.
Limitations
In order to use change streams in MongoDB, there are a few requirements your environment must meet.
The database must be in a replica set or sharded cluster.
The database must use the WiredTiger storage engine.
The replica set or sharded cluster must use replica set protocol version 1.
3. Listener Config
The following parameters should be included in your listener configuration entry when setting up a real time sync using MongoDB.
2.1 Topic Column
Parameter
Value
Database
The name of your MongoDB database.​
Collection
The name of your MongoDB collection.​
Pipeline Stages
Optional. This parameter allows you to specify pipeline stages with filters.

In MongoDB, an aggregation pipeline consists of one or more stages that process documents:
Each stage performs an operation on the input documents. For example, a stage can filter documents, group documents, and calculate values.
The documents that are output from a stage are passed to the next stage.
An aggregation pipeline can return results for groups of documents. For example, return the total, average, maximum, and minimum values.

Our example config uses a filter to return documents with an ID between 0 and 10,000 AND documents with the location set to Montreal, OR where the operation type is 'delete'
Example Config
{
	"database": "cinchy",
	"collection": "employee",
        "changeStreamSettings": {
		"pipelineStages": [
			"{ $match: {
                                       $or: [ 
                                           { $and: [
                                                { 'fullDocument.id': { $gt: 0, $lt: 10000 } }, 
                                                { 'fullDocument.location': 'Montreal' } ] 
                                           }, 
                                           { 'operationType': 'delete' } 
                                      ]
                                } }"
		]
	}
}
​
2.2 Connections Attributes Column
Parameter
Value
Connection String
Your MongoDB connection string.​
Example Config
{
	"connectionString": "mongodb://localhost:9877"
}

Parameter	Value
Database	The name of your MongoDB database.
Collection	The name of your MongoDB collection.
Pipeline Stages	Optional. This parameter allows you to specify pipeline stages with filters. In MongoDB, an aggregation pipeline consists of one or more stages that process documents: Each stage performs an operation on the input documents. For example, a stage can filter documents, group documents, and calculate values. The documents that are output from a stage are passed to the next stage. An aggregation pipeline can return results for groups of documents. For example, return the total, average, maximum, and minimum values. Our example config uses a filter to return documents with an ID between 0 and 10,000 AND documents with the location set to Montreal, OR where the operation type is 'delete'

Parameter	Value
Connection String	Your MongoDB connection string.

Kafka Topic

Salesforce

Last modified 3mo ago