The MongoDB CDC connector allows for reading snapshot data and incremental data from MongoDB. This document describes how to setup the MongoDB CDC connector to run SQL queries against MongoDB.

In order to setup the MongoDB CDC connector, the following table provides dependency information for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.

<dependency>
  <groupId>com.ververica</groupId>
  <artifactId>flink-connector-mongodb-cdc</artifactId>
  <!-- The dependency is available only for stable releases, SNAPSHOT dependencies need to be built based on master or release- branches by yourself. -->
  <version>2.5-SNAPSHOT</version>
</dependency>

Download link is available only for stable releases.

Download flink-sql-connector-mongodb-cdc-2.5-SNAPSHOT.jar and put it under <FLINK_HOME>/lib/.

Note: flink-sql-connector-mongodb-cdc-XXX-SNAPSHOT version is the code corresponding to the development branch. Users need to download the source code and compile the corresponding jar. Users should use the released version, such as flink-sql-connector-mongodb-cdc-2.2.1.jar, the released version will be available in the Maven central warehouse.

• MongoDB version

  MongoDB version >= 3.6
  We use change streams feature (new in version 3.6) to capture change data.

• Cluster Deployment

  replica sets or sharded clusters is required.

• Storage Engine

  WiredTiger storage engine is required.

• Replica set protocol version

  Replica set protocol version 1 (pv1) is required.
  Starting in version 4.0, MongoDB only supports pv1. pv1 is the default for all new replica sets created with MongoDB 3.2 or later.

• Privileges

  changeStream and read privileges are required by MongoDB Kafka Connector.

  You can use the following example for simple authorization.
  For more detailed authorization, please refer to MongoDB Database User Roles.

  use admin;
  db.createRole(
      {
          role: "flinkrole",
          privileges: [{
              // Grant privileges on all non-system collections in all databases
              resource: { db: "", collection: "" },
              actions: [
                  "splitVector",
                  "listDatabases",
                  "listCollections",
                  "collStats",
                  "find",
                  "changeStream" ]
          }],
          roles: [
              // Read config.collections and config.chunks
              // for sharded cluster snapshot splitting.
              { role: 'read', db: 'config' }
          ]
      }
  );
  
  db.createUser(
    {
        user: 'flinkuser',
        pwd: 'flinkpw',
        roles: [
           { role: 'flinkrole', db: 'admin' }
        ]
    }
  );

The MongoDB CDC table can be defined as following:

-- register a MongoDB table 'products' in Flink SQL
CREATE TABLE products (
  _id STRING, // must be declared
  name STRING,
  weight DECIMAL(10,3),
  tags ARRAY<STRING>, -- array
  price ROW<amount DECIMAL(10,2), currency STRING>, -- embedded document
  suppliers ARRAY<ROW<name STRING, address STRING>>, -- embedded documents
  PRIMARY KEY(_id) NOT ENFORCED
) WITH (
  'connector' = 'mongodb-cdc',
  'hosts' = 'localhost:27017,localhost:27018,localhost:27019',
  'username' = 'flinkuser',
  'password' = 'flinkpw',
  'database' = 'inventory',
  'collection' = 'products'
);

-- read snapshot and change events from products collection
SELECT * FROM products;

Note that

MongoDB's change event record doesn't have updated before message. So, we can only convert it to Flink's UPSERT changelog stream. An upsert stream requires a unique key, so we must declare _id as primary key. We can't declare other column as primary key, because delete operation does not contain the key and value besides _id and sharding key.

Note: heartbeat.interval.ms is highly recommended setting a proper value larger than 0 if the collection changes slowly. The heartbeat event can push the resumeToken forward to avoid resumeToken being expired when we recover the Flink job from a checkpoint or savepoint.

The following format metadata can be exposed as read-only (VIRTUAL) columns in a table definition.

The extended CREATE TABLE example demonstrates the syntax for exposing these metadata fields:

CREATE TABLE products (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    collection_name STRING METADATA  FROM 'collection_name' VIRTUAL,
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    _id STRING, // must be declared
    name STRING,
    weight DECIMAL(10,3),
    tags ARRAY<STRING>, -- array
    price ROW<amount DECIMAL(10,2), currency STRING>, -- embedded document
    suppliers ARRAY<ROW<name STRING, address STRING>>, -- embedded documents
    PRIMARY KEY(_id) NOT ENFORCED
) WITH (
    'connector' = 'mongodb-cdc',
    'hosts' = 'localhost:27017,localhost:27018,localhost:27019',
    'username' = 'flinkuser',
    'password' = 'flinkpw',
    'database' = 'inventory',
    'collection' = 'products'
);

The MongoDB CDC connector is a Flink Source connector which will read database snapshot first and then continues to read change stream events with exactly-once processing even failures happen.

The config option scan.startup.mode specifies the startup mode for MongoDB CDC consumer. The valid enumerations are:

• initial (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest oplog.
• latest-offset: Never to perform snapshot on the monitored database tables upon first startup, just read from the end of the oplog which means only have the changes since the connector was started.
• timestamp: Skip snapshot phase and start reading oplog events from a specific timestamp.

For example in DataStream API:

MongoDBSource.builder()
    .startupOptions(StartupOptions.latest()) // Start from latest offset
    .startupOptions(StartupOptions.timestamp(1667232000000L) // Start from timestamp
    .build()

and with SQL:

CREATE TABLE mongodb_source (...) WITH (
    'connector' = 'mongodb-cdc',
    'scan.startup.mode' = 'latest-offset', -- Start from latest offset
    ...
    'scan.startup.mode' = 'timestamp', -- Start from timestamp
    'scan.startup.timestamp-millis' = '1667232000000' -- Timestamp under timestamp startup mode
    ...
)

We integrate the MongoDB's official Kafka Connector to read snapshot or change events from MongoDB and drive it by Debezium's EmbeddedEngine.

Debezium's EmbeddedEngine provides a mechanism for running a single Kafka Connect SourceConnector within an application's process, and it can drive any standard Kafka Connect SourceConnector properly even which is not provided by Debezium.

We choose MongoDB's official Kafka Connector instead of the Debezium's MongoDB Connector because they use a different change data capture mechanism.

• For Debezium's MongoDB Connector, it reads the oplog.rs collection of each replica-set's master node.
• For MongoDB's Kafka Connector, it subscribes Change Stream of MongoDB.

MongoDB's oplog.rs collection doesn't keep the changed record's update before state, so it's hard to extract the full document state by a single oplog.rs record and convert it to change log stream accepted by Flink (Insert Only, Upsert, All). Additionally, MongoDB 5 (released in July 2021) has changed the oplog format, so the current Debezium connector cannot be used with it.

Change Stream is a new feature provided by MongoDB 3.6 for replica sets and sharded clusters that allows applications to access real-time data changes without the complexity and risk of tailing the oplog.
Applications can use change streams to subscribe to all data changes on a single collection, a database, or an entire deployment, and immediately react to them.

Lookup Full Document for Update Operations is a feature provided by Change Stream which can configure the change stream to return the most current majority-committed version of the updated document. Because of this feature, we can easily collect the latest full document and convert the change log to Flink's Upsert Changelog Stream.

By the way, Debezium's MongoDB change streams exploration mentioned by DBZ-435 is on roadmap.
If it's done, we can consider integrating two kinds of source connector for users to choose.

The MongoDB CDC connector can also be a DataStream source. You can create a SourceFunction as the following shows:

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.mongodb.MongoDBSource;

public class MongoDBSourceExample {
    public static void main(String[] args) throws Exception {
        SourceFunction<String> sourceFunction = MongoDBSource.<String>builder()
                .hosts("localhost:27017")
                .username("flink")
                .password("flinkpw")
                .databaseList("inventory") // set captured database, support regex
                .collectionList("inventory.products", "inventory.orders") //set captured collections, support regex
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(sourceFunction)
                .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

        env.execute();
    }
}

The MongoDB CDC incremental connector (after 2.3.0) can be used as the following shows:

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.connectors.mongodb.source.MongoDBSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

public class MongoDBIncrementalSourceExample {
    public static void main(String[] args) throws Exception {
        MongoDBSource<String> mongoSource =
                MongoDBSource.<String>builder()
                        .hosts("localhost:27017")
                        .databaseList("inventory") // set captured database, support regex
                        .collectionList("inventory.products", "inventory.orders") //set captured collections, support regex
                        .username("flink")
                        .password("flinkpw")
                        .deserializer(new JsonDebeziumDeserializationSchema())
                        .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // enable checkpoint
        env.enableCheckpointing(3000);
        // set the source parallelism to 2
        env.fromSource(mongoSource, WatermarkStrategy.noWatermarks(), "MongoDBIncrementalSource")
                .setParallelism(2)
                .print()
                .setParallelism(1);

        env.execute("Print MongoDB Snapshot + Change Stream");
    }
}

Note:

• If database regex is used, readAnyDatabase role is required.
• The incremental snapshot feature only supports after MongoDB 4.0.

MongoDB 6.0 and above supports emitting change stream events containing document before and after the change was made (aka. pre- and post-images).

• The pre-image is the document before it was replaced, updated, or deleted. There is no pre-image for an inserted document.

• The post-image is the document after it was inserted, replaced, or updated. There is no post-image for a deleted document.

MongoDB CDC could make uses of pre-image and post-images to generate full-mode changelog stream including Insert, Update Before, Update After, and Delete data rows, thereby avoiding additional ChangelogNormalize downstream node.

To enable this feature, here's some prerequisites:

• MongoDB version must be 6.0 or above;
• Enable preAndPostImages feature at the database level:

db.runCommand({
  setClusterParameter: {
    changeStreamOptions: {
      preAndPostImages: {
        expireAfterSeconds: 'off' // replace with custom image expiration time
      }
    }
  }
})

• Enable changeStreamPreAndPostImages feature for collections to be monitored:

db.runCommand({
  collMod: "<< collection name >>", 
  changeStreamPreAndPostImages: {
    enabled: true 
  } 
})

• Enable MongoDB CDC's scan.full-changelog feature:

MongoDBSource.builder()
    .scanFullChangelog(true)
    ...
    .build()

or with Flink SQL:

CREATE TABLE mongodb_source (...) WITH (
    'connector' = 'mongodb-cdc',
    'scan.full-changelog' = 'true',
    ...
)

BSON short for Binary JSON is a binary-encoded serialization of JSON-like format used to store documents and make remote procedure calls in MongoDB.

Flink SQL Data Type is similar to the SQL standard’s data type terminology which describes the logical type of a value in the table ecosystem. It can be used to declare input and/or output types of operations.

In order to enable Flink SQL to process data from heterogeneous data sources, the data types of heterogeneous data sources need to be uniformly converted to Flink SQL data types.

The following is the mapping of BSON type and Flink SQL type.

• MongoDB Kafka Connector
• Change Streams
• Replication
• Sharding
• Database User Roles
• WiredTiger
• Replica set protocol
• Connection String Options
• Document Pre- and Post-Images
• BSON Types
• Flink DataTypes

• FAQ(English)
• FAQ(中文)