[ISSUE-393] Add documentation for mongodb-cdc connector #395

Jiabao-Sun · 2021-09-04T18:51:38Z

MongoDB CDC Connector

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.

Dependencies

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.

Maven dependency

<dependency>
  <groupId>com.ververica</groupId>
  <artifactId>flink-connector-mongodb-cdc</artifactId>
  <version>2.1.0</version>
</dependency>

SQL Client JAR

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

Setup MongoDB

Availability

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.createUser({
  user: "flinkuser",
  pwd: "flinkpw",
  roles: [
    { role: "read", db: "admin" }, //read role includes changeStream privilege 
    { role: "readAnyDatabase", db: "admin" } //for snapshot reading
  ]
});

How to create a MongoDB CDC table

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 update 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, becauce delete operation do not contain's the key and value besides _id and sharding key.

Connector Options

Option	Required	Default	Type	Description
connector	required	(none)	String	Specify what connector to use, here should be `'mongodb-cdc'`.
hosts	required	(none)	String	The comma-separated list of hostname and port pairs of the MongoDB servers. eg. localhost:27017,localhost:27018
username	optional	(none)	String	Name of the database user to be used when connecting to MongoDB. This is required only when MongoDB is configured to use authentication.
password	optional	(none)	String	Password to be used when connecting to MongoDB. This is required only when MongoDB is configured to use authentication.
database	required	(none)	String	Name of the database to watch for changes.
collection	required	(none)	String	Name of the collection in the database to watch for changes.
connection.options	optional	(none)	String	The ampersand-separated connection options of MongoDB. eg. `replicaSet=test&connectTimeoutMS=300000`
errors.tolerance	optional	none	String	Whether to continue processing messages if an error is encountered. Accept `none` or `all`. When set to `none`, the connector reports an error and blocks further processing of the rest of the records when it encounters an error. When set to `all`, the connector silently ignores any bad messages.
errors.log.enable	optional	true	Boolean	Whether details of failed operations should be written to the log file.
copy.existing	optional	true	Boolean	Whether copy existing data from source collections.
copy.existing.pipeline	optional	(none)	String	An array of JSON objects describing the pipeline operations to run when copying existing data. This can improve the use of indexes by the copying manager and make copying more efficient. eg. `[{"$match": {"closed": "false"}}]` ensures that only documents in which the closed field is set to false are copied.
copy.existing.max.threads	optional	Processors Count	Integer	The number of threads to use when performing the data copy.
copy.existing.queue.size	optional	16000	Integer	The max size of the queue to use when copying data.
poll.max.batch.size	optional	1000	Integer	Maximum number of change stream documents to include in a single batch when polling for new data.
poll.await.time.ms	optional	1500	Integer	The amount of time to wait before checking for new results on the change stream.
heartbeat.interval.ms	optional	0	Integer	The length of time in milliseconds between sending heartbeat messages. Use 0 to disable.

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

Features

Exactly-Once Processing

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.

Snapshot When Startup Or Not

The config option copy.existing specifies whether do snapshot when MongoDB CDC consumer startup. Defaults to true.

Snapshot Data Filters

The config option copy.existing.pipeline describing the filters when copying existing data.
This can improve the use of indexes by the copying manager and make copying more efficient.

In the following example, the $match aggregation operator ensures that only documents in which the closed field is set to false are copied.
copy.existing.pipeline=[ { "$match": { "closed": "false" } } ]

Change Streams

We integrates 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 cause they use a different change data capture mechanism.

For Debezium's MongoDB Connector, it read the oplog.rs collection of each replicaset's master node.
For MongoDB's Kafka Connector, it subscribe 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) 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 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, 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 support both for users to choose.

DataStream Source

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.StringDebeziumDeserializationSchema;
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")
                .database("inventory")
                .collection("products")
                .deserializer(new StringDebeziumDeserializationSchema())
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

        env.execute();
    }
}

Data Type Mapping

BSON type	Flink SQL type
	TINYINT
	SMALLINT
Int	INT
Long	BIGINT
	FLOAT
Double	DOUBLE
Decimal128	DECIMAL(p, s)
Boolean	BOOLEAN
Date Timestamp	DATE
Date Timestamp	TIME
Date	TIMESTAMP(3) TIMESTAMP_LTZ(3)
Timestamp	TIMESTAMP(0) TIMESTAMP_LTZ(0)
String ObjectId UUID Symbol MD5 JavaScript Regex	STRING
BinData	BYTES
Object	ROW
Array	ARRAY
DBPointer	ROW<$ref STRING, $id STRING>
GeoJSON	Point : ROW<type STRING, coordinates ARRAY<DOUBLE>> Line : ROW<type STRING, coordinates ARRAY<ARRAY< DOUBLE>>> ...

Reference

wuchong · 2021-09-06T07:07:21Z

docs/content/tutorials/tutorial.md

+      - "27017:27017"
+    environment:
+      - MONGO_INITDB_ROOT_USERNAME=mongouser
+      - MONGO_INITDB_ROOT_PASSWORD=mongopw


I have updated the existing tutorial pages to have a better meaningful title. You can create a new page for mongodb, the title can be "Streaming ETL from MongoDB to Elasticsearch"

Thanks @wuchong, I'll move the tutorial to that.

wuchong · 2021-09-06T07:09:07Z

Besides, you can preview docs locally by following the instruction https://github.com/ververica/flink-cdc-connectors/blob/master/docs/README.md

wuchong

Thanks for the work @Jiabao-Sun , I left some comments.

docs/content/connectors/mongodb-cdc.md

wuchong · 2021-09-10T12:17:22Z

docs/content/connectors/mongodb-cdc.md

+This can improve the use of indexes by the copying manager and make copying more efficient.
+
+In the following example, the $match aggregation operator ensures that only documents in which the closed field is set to false are copied.
+`copy.existing.pipeline=[ { "$match": { "closed": "false" } } ]`


Maybe we should quotes $match, this part of the rendering is broken.

wuchong · 2021-09-10T12:19:27Z

docs/content/connectors/mongodb-cdc.md

+- For MongoDB's Kafka Connector, it subscribe `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) has changed the oplog format, so the current Debezium connector cannot be used with it.


released in July 2021.

wuchong · 2021-09-10T12:23:25Z

docs/content/connectors/mongodb-cdc.md

+<table class="colwidths-auto docutils">
+    <thead>
+      <tr>
+        <th class="text-left">BSON type<a href="https://docs.mongodb.com/manual/reference/bson-types/"></a></th>


Explain a bit why we map from BSON type at the front?

wuchong · 2021-09-10T12:25:23Z

docs/content/tutorials/mongodb-tutorial.md

+  postgres:
+    image: debezium/example-postgres:1.1
+    ports:
+      - "5432:5432"
+    environment:
+      - POSTGRES_PASSWORD=1234
+      - POSTGRES_DB=postgres
+      - POSTGRES_USER=postgres
+      - POSTGRES_PASSWORD=postgres
+  mysql:
+    image: debezium/example-mysql:1.1
+    ports:
+      - "3306:3306"
+    environment:
+      - MYSQL_ROOT_PASSWORD=123456
+      - MYSQL_USER=mysqluser
+      - MYSQL_PASSWORD=mysqlpw


Could you remove all content about postgres and mysql because they are not needed?

wuchong · 2021-09-10T12:35:09Z

docs/content/connectors/mongodb-cdc.md

+                .password("flinkpw")
+                .database("inventory")
+                .collection("products")
+                .deserializer(new StringDebeziumDeserializationSchema())


We are going to replace the recommended deserizer to JsonDebeziumDeserializationSchema which is more user-friendly for users. Could you take a look #410 and update this part?

2. Remove all content about postgres and mysql in mongodb-tutorials 3. Brief introduction to type conversion 4. Use JsonDebeziumDeserializationSchema instead of StringDebeziumDeserializationSchema

Jiabao-Sun · 2021-09-12T16:09:50Z

Hi @wuchong, The problem you mentioned above has been fixed.
If you are available, please help review it again.

wuchong

Thanks for the great work @Jiabao-Sun .
LGTM.

Jiabao-Sun mentioned this pull request Sep 4, 2021

[mongodb] Add documentation for newly introduced mongodb-cdc connector #393

Closed

Jiabao-Sun marked this pull request as draft September 5, 2021 05:17

Jiabao-Sun force-pushed the mongodb-cdc-docs branch from 2d76232 to 9388a3c Compare September 5, 2021 17:38

Jiabao-Sun marked this pull request as ready for review September 5, 2021 17:41

wuchong reviewed Sep 6, 2021

View reviewed changes

Jiabao-Sun added 3 commits September 8, 2021 00:21

[doc] Add documentation for mongodb-cdc connector

cf56b6e

[doc] Add tutorial for mongodb-cdc connector

978715e

Change mongodb tutorials to a seperate page

d3a1cd8

Jiabao-Sun force-pushed the mongodb-cdc-docs branch 2 times, most recently from c7257ad to 228b5b1 Compare September 8, 2021 11:16

Use new config options hosts, user, password instead of uri

6fad791

Jiabao-Sun force-pushed the mongodb-cdc-docs branch from 228b5b1 to 6fad791 Compare September 8, 2021 13:19

Jiabao-Sun added 2 commits September 10, 2021 17:05

Add connection.options for MongoDB connection options

4f30b1b

Fix deprecated config user

bca6277

Jiabao-Sun requested a review from wuchong September 10, 2021 10:49

wuchong reviewed Sep 10, 2021

View reviewed changes

1. Add release year about MongoDB 5.0

bd7b36f

2. Remove all content about postgres and mysql in mongodb-tutorials 3. Brief introduction to type conversion 4. Use JsonDebeziumDeserializationSchema instead of StringDebeziumDeserializationSchema

Jiabao-Sun requested a review from wuchong September 10, 2021 17:50

wuchong approved these changes Sep 13, 2021

View reviewed changes

wuchong merged commit f9d17f4 into apache:master Sep 13, 2021

wuchong added this to the v2.1.0 milestone Sep 13, 2021

Jiabao-Sun deleted the mongodb-cdc-docs branch February 17, 2022 15:18

ChaomingZhangCN pushed a commit to ChaomingZhangCN/flink-cdc that referenced this pull request Jan 13, 2025

[mongodb][docs] Add documentation for mongodb-cdc connector (apache#395)

bf8ab1f

[ISSUE-393] Add documentation for mongodb-cdc connector #395

[ISSUE-393] Add documentation for mongodb-cdc connector #395

Uh oh!

Conversation

Jiabao-Sun commented Sep 4, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

MongoDB CDC Connector

Dependencies

Maven dependency

SQL Client JAR

Setup MongoDB

Availability

How to create a MongoDB CDC table

Connector Options

Features

Exactly-Once Processing

Snapshot When Startup Or Not

Snapshot Data Filters

Change Streams

DataStream Source

Data Type Mapping

Reference

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wuchong commented Sep 6, 2021

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Jiabao-Sun commented Sep 12, 2021

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Jiabao-Sun commented Sep 4, 2021 •

edited

Loading