You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2023/02/14 10:38:23 UTC
[flink-connector-mongodb] branch main updated: [FLINK-30967][docs] Add MongoDB connector documentation
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-mongodb.git
The following commit(s) were added to refs/heads/main by this push:
new ae0881d [FLINK-30967][docs] Add MongoDB connector documentation
ae0881d is described below
commit ae0881d56e9a621a08ad85b708162481f850520a
Author: Jiabao Sun <ji...@xtransfer.cn>
AuthorDate: Tue Feb 14 18:38:17 2023 +0800
[FLINK-30967][docs] Add MongoDB connector documentation
---
.../docs/connectors/datastream/mongodb.md | 253 +++++++++++
docs/content.zh/docs/connectors/table/mongodb.md | 443 +++++++++++++++++++
docs/content/docs/connectors/datastream/mongodb.md | 277 ++++++++++++
docs/content/docs/connectors/table/mongodb.md | 474 +++++++++++++++++++++
docs/data/mongodb.yml | 3 +
.../mongodb/source/MongoSourceBuilder.java | 4 +-
.../mongodb/table/MongoConnectorOptions.java | 2 +-
7 files changed, 1453 insertions(+), 3 deletions(-)
diff --git a/docs/content.zh/docs/connectors/datastream/mongodb.md b/docs/content.zh/docs/connectors/datastream/mongodb.md
new file mode 100644
index 0000000..d06bcea
--- /dev/null
+++ b/docs/content.zh/docs/connectors/datastream/mongodb.md
@@ -0,0 +1,253 @@
+---
+title: MongoDB
+weight: 5
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# MongoDB 连接器
+Flink 提供了 [MongoDB](https://www.mongodb.com/) 连接器使用至少一次（At-least-once）的语义在 MongoDB collection 中读取和写入数据。
+
+要使用此连接器，请将以下依赖添加到你的项目中：
+
+{{< connector_artifact flink-connector-mongodb 1.0.0 >}}
+
+## MongoDB Source
+下面的例子介绍了如何构建一个 MongoDB Source：
+
+{{< tabs "a5da31f0-56ed-4c8a-87f6-d02c7867f795" >}}
+{{< tab "Java" >}}
+```java
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.mongodb.source.MongoSource;
+import org.apache.flink.connector.mongodb.source.reader.deserializer.MongoDeserializationSchema;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import org.bson.BsonDocument;
+
+MongoSource<String> source = MongoSource.<String>builder()
+ .setUri("mongodb://user:password@127.0.0.1:27017")
+ .setDatabase("my_db")
+ .setCollection("my_coll")
+ .setProjectedFields("_id", "f0", "f1")
+ .setFetchSize(2048)
+ .setLimit(10000)
+ .setNoCursorTimeout(true)
+ .setPartitionStrategy(PartitionStrategy.SAMPLE)
+ .setPartitionSize(MemorySize.ofMebiBytes(64))
+ .setSamplesPerPartition(10)
+ .setDeserializationSchema(new MongoDeserializationSchema<String>() {
+ @Override
+ public String deserialize(BsonDocument document) {
+ return document.toJson();
+ }
+
+ @Override
+ public TypeInformation<String> getProducedType() {
+ return BasicTypeInfo.STRING_TYPE_INFO;
+ }
+ })
+ .build();
+
+StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "MongoDB-Source")
+ .setParallelism(2)
+ .print()
+ .setParallelism(1);
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+### 配置项
+Flink 的 MongoDB Source 可以通过 `MongoSource.<OutputType>builder()` 构造器构建。
+
+1. __setUri(String uri)__
+ * 必须。
+ * 设置 MongoDB [连接字符串](https://www.mongodb.com/docs/manual/reference/connection-string/)。
+2. __setDatabase(String database)__
+ * 必须。
+ * 设置读取的数据库名称。
+3. __setCollection(String collection)__
+ * 必须。
+ * 设置读取的集合名称。
+4. _setFetchSize(int fetchSize)_
+ * 可选。默认值： `2048`。
+ * 设置每次循环读取时应该从游标中获取的行数。
+5. _setNoCursorTimeout(boolean noCursorTimeout)_
+ * 可选。默认值： `true`。
+ * MongoDB 服务端通常会将空闲时间超过 10 分钟的 cursor 关闭，来节省内存开销。将这个参数设置为 true 可以
+ 防止 cursor 因为读取时间过长或者背压导致的空闲而关闭。需要注意的是。如果 cursor 所在的会话空闲时间超过
+ 了 30 分钟，MongoDB 会关闭当前会话以及由当前会话打开的 cursors，不管有没有设置 `noCursorTimeout()`
+ 或者 `maxTimeMS()` 大于 30分钟。
+6. _setPartitionStrategy(PartitionStrategy partitionStrategy) _
+ * 可选。 默认值： `PartitionStrategy.DEFAULT`。
+ * 设置分区策略。 可选的分区策略有 `SINGLE`，`SAMPLE`，`SPLIT_VECTOR`，`SHARDED` 和 `DEFAULT`。
+ 查看 <a href="#分区策略">分区策略</a> 章节获取详细介绍。
+7. _setPartitionSize(MemorySize partitionSize)_
+ * 可选。默认值：`64mb`。
+ * 设置每个分区的内存大小。通过指定的分区大小，将 MongoDB 的一个集合切分成多个分区。
+ 可以设置并行度，并行地读取这些分区，以提升整体的读取速度。
+8. _setSamplesPerPartition(int samplesPerPartition)_
+ * 可选。默认值：`10`。
+ * 仅用于 `SAMPLE` 抽样分区策略，设置每个分区的样本数量。抽样分区器根据分区键对集合进行随机采样的方式计算分区边界。
+ `总的样本数量 = 每个分区的样本数量 * (文档总数 / 每个分区的文档数量)`
+9. _setLimit(int limit)_
+ * 可选。默认值：`-1`。
+ * 限制每个 reader 最多读取文档的数量。如果我们设置了读取并行度大于 1，那么最多读取的文档数量等于 `并行度 * 限制数量`。
+10. _setProjectedFields(String... projectedFields)_
+ * 可选。
+ * 设置读取文档的部分字段。如果没有设置，会读取文档的全部字段。
+11. __setDeserializationSchema(MongoDeserializationSchema<OutputType> deserializationSchema)__
+ * 必须。
+ * 设置 `MongoDeserializationSchema` 用于解析 MongoDB BSON 类型的文档。
+
+### 分区策略
+使用分区可以利用并行读取来加速整体的读取效率。目前提供了以下几种分区策略：
+
+- `SINGLE`：将整个集合作为一个分区。
+- `SAMPLE`：通过随机采样的方式来生成分区，快速但可能不均匀。
+- `SPLIT_VECTOR`：通过 MongoDB 计算分片的 splitVector 命令来生成分区，快速且均匀。
+ 仅适用于未分片集合，需要 splitVector 权限。
+- `SHARDED`：从 `config.chunks` 集合中直接读取分片集合的分片边界作为分区，不需要额外计算，快速且均匀。
+ 仅适用于已经分片的集合，需要 config 数据库的读取权限。
+- `DEFAULT`：对分片集合使用 `SHARDED` 策略，对未分片集合使用 `SPLIT_VECTOR` 策略。
+
+## MongoDB Sink
+下面的例子介绍了如何构建一个 MongoDB Sink：
+
+{{< tabs "e72bd5ec-6c23-4bc1-85e7-c42290cffa77" >}}
+{{< tab "Java" >}}
+```java
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.mongodb.sink.MongoSink;
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+import com.mongodb.client.model.InsertOneModel;
+import org.bson.BsonDocument;
+
+DataStream<String> stream = ...;
+
+MongoSink<String> sink = MongoSink.<String>builder()
+ .setUri("mongodb://user:password@127.0.0.1:27017")
+ .setDatabase("my_db")
+ .setCollection("my_coll")
+ .setBatchSize(1000)
+ .setBatchIntervalMs(1000)
+ .setMaxRetries(3)
+ .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
+ .setSerializationSchema(
+ (input, context) -> new InsertOneModel<>(BsonDocument.parse(input)))
+ .build();
+
+stream.sinkTo(sink);
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+### 配置项
+Flink 的 MongoDB Source 可以通过 `MongoSink.<InputType>builder()` 构造器构建。
+
+1. __setUri(String uri)__
+ * 必须。
+ * 设置 MongoDB [连接字符串](https://www.mongodb.com/docs/manual/reference/connection-string/)。
+2. __setDatabase(String database)__
+ * 必须。
+ * 设置写入的数据库名称。
+3. __setCollection(String collection)__
+ * 必须。
+ * 设置写入的集合名称。
+4. _setBatchSize(int batchSize)_
+ * 可选。默认值：`1000`。
+ * 设置写入的最大批次大小。可以设置为 -1 来禁用批式写入。
+5. _setBatchIntervalMs(long batchIntervalMs)_
+ * 可选。默认值：`1000`.
+ * 设置写入的最大间隔时间，单位为毫秒。可以设置为 -1 来禁用批式写入。
+6. _setMaxRetries(int maxRetries)_
+ * 可选。默认值：`3`。
+ * 设置写入失败时最大重试次数。
+7. _setDeliveryGuarantee(DeliveryGuarantee deliveryGuarantee)_
+ * 可选。默认值：`DeliveryGuarantee.AT_LEAST_ONCE`.
+ * 设置投递保证。 仅一次（EXACTLY_ONCE）的投递保证暂不支持。
+8. __setSerializationSchema(MongoSerializationSchema<InputType> serializationSchema)__
+ * 必须。
+ * 设置 `MongoSerializationSchema` 将输入类型转换为 MongoDB
+ [WriteModel](https://www.mongodb.com/docs/drivers/java/sync/current/usage-examples/bulkWrite/)。
+
+### 容错
+当开启了 Flink checkpoint，Flink MongoDB Sink 保证至少一次 (at-least-once) 的写入语义.
+`MongoWriter` 在检查点时，会确认所有被缓存的写入操作被正确写入.
+
+关于检查点和容错相关的更多信息，请参阅 [容错文档]({{< ref "docs/learn-flink/fault_tolerance" >}}).
+
+下面的示例介绍了如何开启检查点：
+
+{{< tabs "e13337f3-8e88-4368-af41-850c9fa20cd9" >}}
+{{< tab "Java" >}}
+```java
+final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+env.enableCheckpointing(5000); // checkpoint every 5000 msecs
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+val env = StreamExecutionEnvironment.getExecutionEnvironment()
+env.enableCheckpointing(5000) // checkpoint every 5000 msecs
+```
+
+{{< /tab >}}
+{{< tab "Python" >}}
+
+```python
+env = StreamExecutionEnvironment.get_execution_environment()
+# checkpoint every 5000 msecs
+env.enable_checkpointing(5000)
+```
+
+{{< /tab >}}
+{{< /tabs >}}
+
+<p style="border-radius: 5px; padding: 5px" class="bg-info">
+<b>重要提醒</b>: 尽管默认的写入语义为至少一次 (AT_LEAST_ONCE)，但检查点并不是默认开启的.
+这会导致 Sink 缓冲写入请求，直到完成或 `MongoWriter` 自动刷新.
+默认情况下，`MongoWriter` 会缓冲 1000 个新增的写入请求。需要使写入更加频繁，请参考 <a href="#配置-Mongo-Writer">配置 MongoWriter</a>。
+</p>
+
+<p style="border-radius: 5px; padding: 5px" class="bg-info">
+当开启 AT_LEAST_ONCE 投递时，通过指定明确的主键和 upsert 方式写入，可以实现仅一次 (exactly-once) 的写入语义。
+</p>
+
+### 配置 Mongo Writer
+
+内部的 `MongoWriter` 可以使用 `MongoSinkBuilder` 更精细化地配置来实现不一样的写入行为：
+
+- **setBatchSize(int batchSize)**：设置写入的批次大小。 可以设置为 -1 来禁用批式写入。
+- **setBatchIntervalMs(long batchIntervalMs)**：设置写入的最大间隔时间，单位为毫秒。 可以设置为 -1 来禁用批式写入。
+
+当使用如下设置时，会产生不一样的写入行为：
+
+- 当缓存记录数超过最大批次大小，或者写入时间间隔超过限制时写入。
+ * batchSize > 1 and batchInterval > 0
+- 仅在检查点写入。
+ * batchSize == -1 and batchInterval == -1
+- 对每条记录进行写入。
+ * batchSize == 1 or batchInterval == 0
diff --git a/docs/content.zh/docs/connectors/table/mongodb.md b/docs/content.zh/docs/connectors/table/mongodb.md
new file mode 100644
index 0000000..3b6d899
--- /dev/null
+++ b/docs/content.zh/docs/connectors/table/mongodb.md
@@ -0,0 +1,443 @@
+---
+title: MongoDB
+weight: 5
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# MongoDB SQL 连接器
+
+{{< label "Scan Source: Bounded" >}}
+{{< label "Lookup Source: Sync Mode" >}}
+{{< label "Sink: Batch" >}}
+{{< label "Sink: Streaming Append & Upsert Mode" >}}
+
+MongoDB 连接器提供了从 MongoDB 中读取和写入数据的能力。
+本文档介绍如何设置 MongoDB 连接器以对 MongoDB 运行 SQL 查询。
+
+连接器可以在 upsert 模式下运行，使用在 DDL 上定义的主键与外部系统交换 UPDATE/DELETE 消息。
+
+如果 DDL 上没有定义主键，则连接器只能以 append 模式与外部系统交换 INSERT 消息且不支持消费 UPDATE/DELETE 消息。
+
+依赖
+------------
+In order to use the MongoDB connector the following dependencies are required for both projects
+using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.
+
+{{< sql_connector_download_table "mongodb" 1.0.0 >}}
+
+MongoDB 连接器目前并不包含在 Flink 的二进制发行版中，请查阅[这里]({{< ref "docs/dev/configuration/overview" >}})了解如何在集群运行中引用 MongoDB 连接器。
+
+如何创建 MongoDB 表
+----------------
+
+MongoDB 表可以以如下方式定义:
+
+```sql
+-- 在 Flink SQL 中注册一张 MongoDB 'users' 表
+CREATE TABLE MyUserTable (
+ _id STRING,
+ name STRING,
+ age INT,
+ status BOOLEAN,
+ PRIMARY KEY (_id) NOT ENFORCED
+) WITH (
+ 'connector' = 'mongodb',
+ 'uri' = 'mongodb://user:password@127.0.0.1:27017',
+ 'database' = 'my_db',
+ 'collection' = 'users'
+);
+
+-- 读取表 "T" 的数据并写入至 MongoDB 表
+INSERT INTO MyUserTable
+SELECT _id, name, age, status FROM T;
+
+-- 从 MongoDB 表中读取数据
+SELECT id, name, age, status FROM MyUserTable;
+
+-- 将 MongoDB 表作为维度表关联
+SELECT * FROM myTopic
+LEFT JOIN MyUserTable FOR SYSTEM_TIME AS OF myTopic.proctime
+ON myTopic.key = MyUserTable._id;
+```
+
+连接器参数
+----------------
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">参数</th>
+ <th class="text-left" style="width: 8%">是否必选</th>
+ <th class="text-left" style="width: 8%">是否透传</th>
+ <th class="text-left" style="width: 7%">默认值</th>
+ <th class="text-left" style="width: 10%">数据类型</th>
+ <th class="text-left" style="width: 42%">描述</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>connector</h5></td>
+ <td>必填</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>指定使用什么类型的连接器，这里应该是<code>'mongodb'</code>。</td>
+ </tr>
+ <tr>
+ <td><h5>uri</h5></td>
+ <td>必填</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>MongoDB 连接字符串。</td>
+ </tr>
+ <tr>
+ <td><h5>database</h5></td>
+ <td>必填</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>读取或写入的数据库。</td>
+ </tr>
+ <tr>
+ <td><h5>collection</h5></td>
+ <td>必填</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>读取或写入的集合。</td>
+ </tr>
+ <tr>
+ <td><h5>scan.fetch-size</h5></td>
+ <td>可选</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">2048</td>
+ <td>Integer</td>
+ <td>每次循环读取时应该从游标中获取的行数。</td>
+ </tr>
+ <tr>
+ <td><h5>scan.cursor.no-timeout</h5></td>
+ <td>可选</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>MongoDB 服务端通常会将空闲时间超过 10 分钟的 cursor 关闭，来节省内存开销。将这个参数设置为 true 可以
+ 防止 cursor 因为读取时间过长或者背压导致的空闲而关闭。需要注意的是。如果 cursor 所在的会话空闲时间超过
+ 了 30 分钟，MongoDB 会关闭当前会话以及由当前会话打开的 cursors，不管有没有设置 `noCursorTimeout()`
+ 或者 `maxTimeMS()` 大于 30分钟。</td>
+ </tr>
+ <tr>
+ <td><h5>scan.partition.strategy</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">default</td>
+ <td>String</td>
+ <td>设置分区策略。可以选择的分区策略有`single`，`sample`，`split-vector`，`sharded` 和 `default`。
+ 请参阅<a href="#分区扫描">分区扫描</a>部分了解更多详情。</td>
+ </tr>
+ <tr>
+ <td><h5>scan.partition.size</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">64mb</td>
+ <td>MemorySize</td>
+ <td>设置每个分区的内存大小。通过指定的分区大小，将 MongoDB 的一个集合切分成多个分区。</td>
+ </tr>
+ <tr>
+ <td><h5>scan.partition.samples</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">10</td>
+ <td>Integer</td>
+ <td>仅用于 `SAMPLE` 抽样分区策略，设置每个分区的样本数量。抽样分区器根据分区键对集合进行随机采样的方式计算分区边界。
+ `总的样本数量 = 每个分区的样本数量 * (文档总数 / 每个分区的文档数量)`。</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.cache</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">NONE</td>
+ <td><p>枚举类型</p>可选值: NONE, PARTIAL</td>
+ <td>维表的缓存策略。 目前支持 NONE（不缓存）和 PARTIAL（只在外部数据库中查找数据时缓存）。</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.partial-cache.max-rows</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Long</td>
+ <td>维表缓存的最大行数，若超过该值，则最老的行记录将会过期。
+ 使用该配置时 "lookup.cache" 必须设置为 "PARTIAL”。请参阅下面的 <a href="#lookup-cache">Lookup Cache</a> 部分了解更多详情。</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.partial-cache.expire-after-write</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>在记录写入缓存后该记录的最大保留时间。使用该配置时 "lookup.cache" 必须设置为 "PARTIAL”。
+ 请参阅下面的 <a href="#lookup-cache">Lookup Cache</a> 部分了解更多详情。</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.partial-cache.expire-after-access</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>在缓存中的记录被访问后该记录的最大保留时间。
+ 使用该配置时 "lookup.cache" 必须设置为 "PARTIAL”。
+ 请参阅下面的 <a href="#lookup-cache">Lookup Cache</a> 部分了解更多详情。</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.partial-cache.caching-missing-key</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>是否缓存维表中不存在的键，默认为true。
+ 使用该配置时 "lookup.cache" 必须设置为 "PARTIAL”。</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.max-retries</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">3</td>
+ <td>Integer</td>
+ <td>查询数据库失败的最大重试次数。</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.retry.interval</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">1s</td>
+ <td>Duration</td>
+ <td>查询数据库失败的最大重试时间。</td>
+ </tr>
+ <tr>
+ <td><h5>sink.buffer-flush.max-rows</h5></td>
+ <td>可选</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">1000</td>
+ <td>Integer</td>
+ <td>设置写入的最大批次大小。</td>
+ </tr>
+ <tr>
+ <td><h5>sink.buffer-flush.interval</h5></td>
+ <td>可选</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">1s</td>
+ <td>Duration</td>
+ <td>设置写入的最大间隔时间。</td>
+ </tr>
+ <tr>
+ <td><h5>sink.max-retries</h5></td>
+ <td>可选</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">3</td>
+ <td>Integer</td>
+ <td>设置写入失败时最大重试次数。</td>
+ </tr>
+ <tr>
+ <td><h5>sink.retry.interval</h5></td>
+ <td>可选</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">1s</td>
+ <td>Duration</td>
+ <td>设置写入失败时最大重试时间。</td>
+ </tr>
+ <tr>
+ <td><h5>sink.parallelism</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>用于定义 MongoDB sink 算子的并行度。默认情况下，并行度是由框架决定：使用与上游链式算子相同的并行度。</td>
+ </tr>
+ <tr>
+ <td><h5>sink.delivery-guarantee</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">at-lease-once</td>
+ <td><p>Enum</p>可选值: none, at-least-once</td>
+ <td>设置投递保证。仅一次（exactly-once）的投递保证暂不支持。</td>
+ </tr>
+ </tbody>
+</table>
+
+特性
+--------
+
+### 键处理
+当写入数据到外部数据库时，Flink 会使用 DDL 中定义的主键。
+如果定义了主键，则连接器将以 upsert 模式工作，否则连接器将以 append 模式工作。
+
+在 MongoDB 连接器中主键用来计算 MongoDB 文档的保留主键 _id。
+MongoDB 的文档 _id 是唯一的且不可变的，可以是除了 Array 外的任何 [BSON 类型](https://www.mongodb.com/docs/manual/reference/bson-types/#bson-types)。
+如果 _id 是嵌套的，嵌套字段的名称不能包含 ($) 符号。
+
+MongoDB 的主键有一些限制。
+在 MongoDB 4.2 之前，索引值的大小需要小于 1024 字节。
+在 MongoDB 4.2 及之后版本，移除了索引大小的限制。
+请参阅 [Index Key Limit](https://www.mongodb.com/docs/manual/reference/limits/#mongodb-limit-Index-Key-Limit) 了解更多详情。
+
+MongoDB 连接器通过将 DDL 声明的主键进行组合，来生成文档的保留主键 _id。
+
+- 当 DDL 中声明的主键仅包含一个字段时，将字段数据转换为 bson 类型的值作为相应文档的_id。
+- 当 DDL 中声明的主键包含多个字段时，我们将所有声明的主键字段组合为一个 bson document 的嵌套类型，作为相应文档的_id。
+ 举个例子，如果我们在 DDL 中定义主键为 `PRIMARY KEY (f1, f2) NOT ENFORCED`，那么相应文档的 _id 将会是 `_id: {f1: v1, f2: v2}` 的形式。
+
+需要注意的是：当 _id 字段存在于 DDL 中，但是主键并没有指定为 _id 时，会产生歧义。
+要么将 _id 字段声明为主键，或者将 _id 字段进行重命名。
+
+有关 PRIMARY KEY 语法的更多详细信息，请参见 [CREATE TABLE DDL]({{< ref "docs/dev/table/sql/create" >}}#create-table)。
+
+### 分区扫描
+为了在并行 `Source` task 实例中加速读取数据，Flink 为 MongoDB table 提供了分区扫描的特性。
+
+- `single`: 将整个集合作为一个分区。
+- `sample`: 通过随机采样的方式来生成分区，快速但可能不均匀。
+- `split-vector`: 通过 MongoDB 计算分片的 splitVector 命令来生成分区，快速且均匀。
+ 仅适用于未分片集合，需要 splitVector 权限。
+- `sharded`: 从 `config.chunks` 集合中直接读取分片集合的分片边界作为分区，不需要额外计算，快速且均匀。
+ 仅适用于已经分片的集合，需要 config 数据库的读取权限。
+- `default`: 对分片集合使用 `sharded` 策略，对未分片集合使用 `split-vector` 策略。
+
+### Lookup Cache
+
+MongoDB 连接器可以用在时态表关联中作为一个可 lookup 的 source (又称为维表)，当前只支持同步的查找模式。
+
+默认情况下，lookup cache 是未启用的，你可以将 `lookup.cache` 设置为 `PARTIAL` 参数来启用。
+
+lookup cache 的主要目的是用于提高时态表关联 MongoDB 连接器的性能。默认情况下，lookup cache 不开启，所以所有请求都会发送到外部数据库。
+当 lookup cache 被启用时，每个进程（即 TaskManager）将维护一个缓存。Flink 将优先查找缓存，只有当缓存未查找到时才向外部数据库发送请求，并使用返回的数据更新缓存。
+当缓存命中最大缓存行 `lookup.partial-cache.max-rows` 或当行超过 `lookup.partial-cache.expire-after-write` 或 `lookup.partial-cache.expire-after-access` 指定的最大存活时间时，缓存中的行将被设置为已过期。
+缓存中的记录可能不是最新的，用户可以将缓存记录超时设置为一个更小的值以获得更好的刷新数据，但这可能会增加发送到数据库的请求数。所以要做好吞吐量和正确性之间的平衡。
+
+默认情况下，flink 会缓存主键的空查询结果，你可以通过将 `lookup.partial-cache.caching-missing-key` 设置为 false 来切换行为。
+
+### 幂等写入
+如果在 DDL 中定义了主键，MongoDB sink 将使用 upsert 语义而不是普通的 INSERT 语句。
+我们将 DDL 中声明的主键进行组合作为 MongoDB 保留主键 _id，使用 upsert 模式进行写入，来保证写入的幂等性。
+
+如果出现故障，Flink 作业会从上次成功的 checkpoint 恢复并重新处理，这可能导致在恢复过程中重复处理消息。
+强烈推荐使用 upsert 模式，因为如果需要重复处理记录，它有助于避免违反数据库主键约束和产生重复数据。
+
+数据类型映射
+----------------
+MongoDB BSON 类型到 Flink SQL 数据类型的映射如下表所示。
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left"><a href="https://www.mongodb.com/docs/manual/reference/bson-types/#bson-types">MongoDB BSON type</a></th>
+ <th class="text-left"><a href="{{< ref "docs/dev/table/types" >}}">Flink SQL type</a></th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><code>ObjectId</code></td>
+ <td><code>STRING</code></td>
+ </tr>
+ <tr>
+ <td><code>String</code></td>
+ <td><code>STRING</code></td>
+ </tr>
+ <tr>
+ <td><code>Boolean</code></td>
+ <td><code>BOOLEAN</code></td>
+ </tr>
+ <tr>
+ <td><code>Binary</code></td>
+ <td><code>BINARY</code><br>
+ <code>VARBINARY</code></td>
+ </tr>
+ <tr>
+ <td><code>Int32</code></td>
+ <td><code>INTEGER</code></td>
+ </tr>
+ <tr>
+ <td><code>-</code></td>
+ <td><code>TINYINT</code><br>
+ <code>SMALLINT</code><br>
+ <code>FLOAT</code></td>
+ </tr>
+ <tr>
+ <td><code>Int64</code></td>
+ <td><code>BIGINT</code></td>
+ </tr>
+ <tr>
+ <td><code>Double</code></td>
+ <td><code>DOUBLE</code></td>
+ </tr>
+ <tr>
+ <td><code>Decimal128</code></td>
+ <td><code>DECIMAL</code></td>
+ </tr>
+ <tr>
+ <td><code>DateTime</code></td>
+ <td><code>TIMESTAMP_LTZ(3)</code></td>
+ </tr>
+ <tr>
+ <td><code>Timestamp</code></td>
+ <td><code>TIMESTAMP_LTZ(0)</code></td>
+ </tr>
+ <tr>
+ <td><code>Object</code></td>
+ <td><code>ROW</code></td>
+ </tr>
+ <tr>
+ <td><code>Array</code></td>
+ <td><code>ARRAY</code></td>
+ </tr>
+ </tbody>
+</table>
+
+对于 MongoDB 独有的类型，我们使用 [Extended JSON 格式](https://www.mongodb.com/docs/drivers/java/sync/current/fundamentals/data-formats/document-data-format-extended-json/)
+将其映射为 Flink SQL STRING 类型。
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left"><a href="https://www.mongodb.com/docs/manual/reference/bson-types/#bson-types">MongoDB BSON type</a></th>
+ <th class="text-left">Flink SQL STRING</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><code>Symbol</code></td>
+ <td><code>{"_value": {"$symbol": "12"}}</code></td>
+ </tr>
+ <tr>
+ <td><code>RegularExpression</code></td>
+ <td><code>{"_value": {"$regularExpression": {"pattern": "^9$", "options": "i"}}}</code></td>
+ </tr>
+ <tr>
+ <td><code>JavaScript</code></td>
+ <td><code>{"_value": {"$code": "function() { return 10; }"}}</code></td>
+ </tr>
+ <tr>
+ <td><code>DbPointer</code></td>
+ <td><code>{"_value": {"$dbPointer": {"$ref": "db.coll", "$id": {"$oid": "63932a00da01604af329e33c"}}}}</code></td>
+ </tr>
+ </tbody>
+</table>
+
+{{< top >}}
diff --git a/docs/content/docs/connectors/datastream/mongodb.md b/docs/content/docs/connectors/datastream/mongodb.md
new file mode 100644
index 0000000..742eb3d
--- /dev/null
+++ b/docs/content/docs/connectors/datastream/mongodb.md
@@ -0,0 +1,277 @@
+---
+title: MongoDB
+weight: 5
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# MongoDB Connector
+Flink provides a [MongoDB](https://www.mongodb.com/) connector for reading and writing data from
+and to MongoDB collections with at-least-once guarantees.
+
+To use this connector, add one of the following dependencies to your project.
+
+{{< connector_artifact flink-connector-mongodb 1.0.0 >}}
+
+## MongoDB Source
+
+The example below shows how to configure and create a source:
+
+{{< tabs "d954f0e8-576a-4152-bfe7-b70bf46c34d2" >}}
+{{< tab "Java" >}}
+```java
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.mongodb.source.MongoSource;
+import org.apache.flink.connector.mongodb.source.reader.deserializer.MongoDeserializationSchema;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import org.bson.BsonDocument;
+
+MongoSource<String> source = MongoSource.<String>builder()
+ .setUri("mongodb://user:password@127.0.0.1:27017")
+ .setDatabase("my_db")
+ .setCollection("my_coll")
+ .setProjectedFields("_id", "f0", "f1")
+ .setFetchSize(2048)
+ .setLimit(10000)
+ .setNoCursorTimeout(true)
+ .setPartitionStrategy(PartitionStrategy.SAMPLE)
+ .setPartitionSize(MemorySize.ofMebiBytes(64))
+ .setSamplesPerPartition(10)
+ .setDeserializationSchema(new MongoDeserializationSchema<String>() {
+ @Override
+ public String deserialize(BsonDocument document) {
+ return document.toJson();
+ }
+
+ @Override
+ public TypeInformation<String> getProducedType() {
+ return BasicTypeInfo.STRING_TYPE_INFO;
+ }
+ })
+ .build();
+
+StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "MongoDB-Source")
+ .setParallelism(2)
+ .print()
+ .setParallelism(1);
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+### Configurations
+Flink's MongoDB source is created by using the static builder `MongoSource.<OutputType>builder()`.
+
+1. __setUri(String uri)__
+ * Required.
+ * Sets the [connection string](https://www.mongodb.com/docs/manual/reference/connection-string/) of MongoDB.
+2. __setDatabase(String database)__
+ * Required.
+ * Name of the database to read from.
+3. __setCollection(String collection)__
+ * Required.
+ * Name of the collection to read from.
+4. _setFetchSize(int fetchSize)_
+ * Optional. Default: `2048`.
+ * Sets the number of documents should be fetched per round-trip when reading.
+5. _setNoCursorTimeout(boolean noCursorTimeout)_
+ * Optional. Default: `true`.
+ * The MongoDB server normally times out idle cursors after an inactivity period (10 minutes) to
+ prevent excess memory use. Set this option to prevent that. If a session is idle for longer
+ than 30 minutes, the MongoDB server marks that session as expired and may close it at any
+ time. When the MongoDB server closes the session, it also kills any in-progress operations
+ and open cursors associated with the session. This includes cursors configured with
+ `noCursorTimeout()` or a `maxTimeMS()` greater than 30 minutes.
+6. _setPartitionStrategy(PartitionStrategy partitionStrategy)_
+ * Optional. Default: `PartitionStrategy.DEFAULT`.
+ * Sets the partition strategy. Available partition strategies are `SINGLE`, `SAMPLE`, `SPLIT_VECTOR`,
+ `SHARDED` and `DEFAULT`. You can see <a href="#partition-strategies">Partition Strategies section</a> for detail.
+7. _setPartitionSize(MemorySize partitionSize)_
+ * Optional. Default: `64mb`.
+ * Sets the partition memory size of MongoDB split. Split a MongoDB collection into multiple
+ partitions according to the partition memory size. Partitions can be read in parallel by
+ multiple readers to speed up the overall read time.
+8. _setSamplesPerPartition(int samplesPerPartition)_
+ * Optional. Default: `10`.
+ * Sets the number of samples to take per partition which is only used for the sample partition
+ strategy `SAMPLE`. The sample partitioner samples the collection, projects and sorts by the
+ partition fields. Then uses every `samplesPerPartition` as the value to use to calculate the
+ partition boundaries. The total number of samples taken is:
+ `samples per partition * ( count of documents / number of documents per partition)`.
+9. _setLimit(int limit)_
+ * Optional. Default: `-1`.
+ * Sets the limit of documents for each reader to read. If limit is not set or set to -1,
+ the documents of the entire collection will be read. If we set the parallelism of reading to
+ be greater than 1, the maximum documents to read is equal to the `parallelism * limit`.
+10. _setProjectedFields(String... projectedFields)_
+ * Optional.
+ * Sets the projection fields of documents to read. If projected fields is not set, all fields of
+ the collection will be read.
+11. __setDeserializationSchema(MongoDeserializationSchema<OutputType> deserializationSchema)__
+ * Required.
+ * A `MongoDeserializationSchema` is required for parsing MongoDB BSON documents.
+
+### Partition Strategies
+Partitions can be read in parallel by multiple readers to speed up the overall read time.
+The following partition strategies are provided:
+
+- `SINGLE`: treats the entire collection as a single partition.
+- `SAMPLE`: samples the collection and generate partitions which is fast but possibly uneven.
+- `SPLIT_VECTOR`: uses the splitVector command to generate partitions for non-sharded
+ collections which is fast and even. The splitVector permission is required.
+- `SHARDED`: reads `config.chunks` (MongoDB splits a sharded collection into chunks, and the
+ range of the chunks are stored within the collection) as the partitions directly. The
+ sharded strategy only used for sharded collection which is fast and even. Read permission
+ of config database is required.
+- `DEFAULT`: uses sharded strategy for sharded collections otherwise using split vector
+ strategy.
+
+## MongoDB Sink
+The example below shows how to configure and create a sink:
+
+{{< tabs "4a96142b-57e4-490f-897c-59ed12c2caef" >}}
+{{< tab "Java" >}}
+```java
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.mongodb.sink.MongoSink;
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+import com.mongodb.client.model.InsertOneModel;
+import org.bson.BsonDocument;
+
+DataStream<String> stream = ...;
+
+MongoSink<String> sink = MongoSink.<String>builder()
+ .setUri("mongodb://user:password@127.0.0.1:27017")
+ .setDatabase("my_db")
+ .setCollection("my_coll")
+ .setBatchSize(1000)
+ .setBatchIntervalMs(1000)
+ .setMaxRetries(3)
+ .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
+ .setSerializationSchema(
+ (input, context) -> new InsertOneModel<>(BsonDocument.parse(input)))
+ .build();
+
+stream.sinkTo(sink);
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+### Configurations
+Flink's MongoDB sink is created by using the static builder `MongoSink.<InputType>builder()`.
+
+1. __setUri(String uri)__
+ * Required.
+ * Sets the [connection string](https://www.mongodb.com/docs/manual/reference/connection-string/) of MongoDB.
+2. __setDatabase(String database)__
+ * Required.
+ * Name of the database to sink to.
+3. __setCollection(String collection)__
+ * Required.
+ * Name of the collection to sink to.
+4. _setBatchSize(int batchSize)_
+ * Optional. Default: `1000`.
+ * Sets the maximum number of actions to buffer for each batch request.
+ You can pass -1 to disable batching.
+5. _setBatchIntervalMs(long batchIntervalMs)_
+ * Optional. Default: `1000`.
+ * Sets the batch flush interval, in milliseconds. You can pass -1 to disable it.
+6. _setMaxRetries(int maxRetries)_
+ * Optional. Default: `3`.
+ * Sets the max retry times if writing records failed.
+7. _setDeliveryGuarantee(DeliveryGuarantee deliveryGuarantee)_
+ * Optional. Default: `DeliveryGuarantee.AT_LEAST_ONCE`.
+ * Sets the wanted `DeliveryGuarantee`. The `EXACTLY_ONCE` guarantee is not supported yet.
+8. __setSerializationSchema(MongoSerializationSchema<InputType> serializationSchema)__
+ * Required.
+ * A `MongoSerializationSchema` is required for parsing input record to MongoDB
+ [WriteModel](https://www.mongodb.com/docs/drivers/java/sync/current/usage-examples/bulkWrite/).
+
+### Fault Tolerance
+
+With Flink’s checkpointing enabled, the Flink MongoDB Sink guarantees
+at-least-once delivery of write operations to MongoDB clusters. It does
+so by waiting for all pending write operations in the `MongoWriter` at the
+time of checkpoints. This effectively assures that all requests before the
+checkpoint was triggered have been successfully acknowledged by MongoDB, before
+proceeding to process more records sent to the sink.
+
+More details on checkpoints and fault tolerance are in the [fault tolerance docs]({{< ref "docs/learn-flink/fault_tolerance" >}}).
+
+To use fault tolerant MongoDB Sinks, checkpointing of the topology needs to be enabled at the execution environment:
+
+{{< tabs "aae2c01d-8f80-4f5f-a707-333607f35669" >}}
+{{< tab "Java" >}}
+```java
+final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+env.enableCheckpointing(5000); // checkpoint every 5000 msecs
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+val env = StreamExecutionEnvironment.getExecutionEnvironment()
+env.enableCheckpointing(5000) // checkpoint every 5000 msecs
+```
+
+{{< /tab >}}
+{{< tab "Python" >}}
+
+```python
+env = StreamExecutionEnvironment.get_execution_environment()
+# checkpoint every 5000 msecs
+env.enable_checkpointing(5000)
+```
+
+{{< /tab >}}
+{{< /tabs >}}
+
+<p style="border-radius: 5px; padding: 5px" class="bg-info">
+<b>IMPORTANT</b>: Checkpointing is not enabled by default but the default delivery guarantee is AT_LEAST_ONCE.
+This causes the sink to buffer requests until it either finishes or the `MongoWriter` flushes automatically.
+By default, the `MongoWriter` will flush after 1000 added write operations. To configure the writer to flush more frequently,
+please refer to the <a href="#configuring-the-internal-mongo-writer">MongoWriter configuration section</a>.
+</p>
+
+<p style="border-radius: 5px; padding: 5px" class="bg-info">
+Using WriteModel with deterministic ids and the upsert method it is possible to achieve exactly-once
+semantics in MongoDB when AT_LEAST_ONCE delivery is configured for the connector.
+</p>
+
+### Configuring the Internal Mongo Writer
+
+The internal `MongoWriter` can be further configured for its behaviour on how write operations are
+flushed, by using the following methods of the `MongoSinkBuilder`:
+
+- **setBatchSize(int batchSize)**: Maximum amount of write operations to buffer before flushing. You can pass -1 to disable it.
+- **setBatchIntervalMs(long batchIntervalMs)**: Interval at which to flush regardless of the size of buffered write operations. You can pass -1 to disable it.
+
+When set as follows, there are the following writing behaviours:
+
+- Flush when time interval or batch size exceed limit.
+ * batchSize > 1 and batchInterval > 0
+- Flush only on checkpoint.
+ * batchSize == -1 and batchInterval == -1
+- Flush for every single write operation.
+ * batchSize == 1 or batchInterval == 0
diff --git a/docs/content/docs/connectors/table/mongodb.md b/docs/content/docs/connectors/table/mongodb.md
new file mode 100644
index 0000000..e5d2387
--- /dev/null
+++ b/docs/content/docs/connectors/table/mongodb.md
@@ -0,0 +1,474 @@
+---
+title: MongoDB
+weight: 5
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# MongoDB SQL Connector
+
+{{< label "Scan Source: Bounded" >}}
+{{< label "Lookup Source: Sync Mode" >}}
+{{< label "Sink: Batch" >}}
+{{< label "Sink: Streaming Append & Upsert Mode" >}}
+
+The MongoDB connector allows for reading data from and writing data into MongoDB.
+This document describes how to set up the MongoDB connector to run SQL queries against MongoDB.
+
+The connector can operate in upsert mode for exchanging UPDATE/DELETE messages with the external
+system using the primary key defined on the DDL.
+
+If no primary key is defined on the DDL, the connector can only operate in append mode for
+exchanging INSERT only messages with external system.
+
+Dependencies
+------------
+
+In order to use the MongoDB connector the following dependencies are required for both projects
+using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.
+
+{{< sql_connector_download_table "mongodb" 1.0.0 >}}
+
+The MongoDB connector is not part of the binary distribution.
+See how to link with it for cluster execution [here]({{< ref "docs/dev/configuration/overview" >}}).
+
+How to create a MongoDB table
+----------------
+
+The MongoDB table can be defined as following:
+
+```sql
+-- register a MongoDB table 'users' in Flink SQL
+CREATE TABLE MyUserTable (
+ _id STRING,
+ name STRING,
+ age INT,
+ status BOOLEAN,
+ PRIMARY KEY (_id) NOT ENFORCED
+) WITH (
+ 'connector' = 'mongodb',
+ 'uri' = 'mongodb://user:password@127.0.0.1:27017',
+ 'database' = 'my_db',
+ 'collection' = 'users'
+);
+
+-- write data into the MongoDB table from the other table "T"
+INSERT INTO MyUserTable
+SELECT _id, name, age, status FROM T;
+
+-- scan data from the MongoDB table
+SELECT id, name, age, status FROM MyUserTable;
+
+-- temporal join the MongoDB table as a dimension table
+SELECT * FROM myTopic
+LEFT JOIN MyUserTable FOR SYSTEM_TIME AS OF myTopic.proctime
+ON myTopic.key = MyUserTable._id;
+```
+
+Connector Options
+----------------
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Option</th>
+ <th class="text-left" style="width: 8%">Required</th>
+ <th class="text-left" style="width: 8%">Forwarded</th>
+ <th class="text-left" style="width: 7%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 42%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>connector</h5></td>
+ <td>required</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Specify what connector to use, here should be <code>'mongodb'</code>.</td>
+ </tr>
+ <tr>
+ <td><h5>uri</h5></td>
+ <td>required</td>
+ <td>yes</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The MongoDB connection uri.</td>
+ </tr>
+ <tr>
+ <td><h5>database</h5></td>
+ <td>required</td>
+ <td>yes</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The name of MongoDB database to read or write.</td>
+ </tr>
+ <tr>
+ <td><h5>collection</h5></td>
+ <td>required</td>
+ <td>yes</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The name of MongoDB collection to read or write.</td>
+ </tr>
+ <tr>
+ <td><h5>scan.fetch-size</h5></td>
+ <td>optional</td>
+ <td>yes</td>
+ <td style="word-wrap: break-word;">2048</td>
+ <td>Integer</td>
+ <td>Gives the reader a hint as to the number of documents that should be fetched from the database per round-trip when reading.</td>
+ </tr>
+ <tr>
+ <td><h5>scan.cursor.no-timeout</h5></td>
+ <td>optional</td>
+ <td>yes</td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>MongoDB server normally times out idle cursors after an inactivity period (10 minutes)
+ to prevent excess memory use. Set this option to true to prevent that. However, if the
+ application takes longer than 30 minutes to process the current batch of documents,
+ the session is marked as expired and closed.</td>
+ </tr>
+ <tr>
+ <td><h5>scan.partition.strategy</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">default</td>
+ <td>String</td>
+ <td>Specifies the partition strategy. Available strategies are `single`, `sample`, `split-vector`, `sharded` and `default`.
+ See the following <a href="#partitioned-scan">Partitioned Scan</a> section for more details.</td>
+ </tr>
+ <tr>
+ <td><h5>scan.partition.size</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">64mb</td>
+ <td>MemorySize</td>
+ <td>Specifies the partition memory size.</td>
+ </tr>
+ <tr>
+ <td><h5>scan.partition.samples</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">10</td>
+ <td>Integer</td>
+ <td>Specifies the samples count per partition. It only takes effect when the partition strategy is sample.
+ The sample partitioner samples the collection, projects and sorts by the partition fields.
+ Then uses every `scan.partition.samples` as the value to use to calculate the partition boundaries.
+ The total number of samples taken is calculated as: `samples per partition * (count of documents / number of documents per partition)`.</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.cache</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">NONE</td>
+ <td><p>Enum</p>Possible values: NONE, PARTIAL</td>
+ <td>The cache strategy for the lookup table. Currently supports NONE (no caching) and PARTIAL (caching entries on lookup operation in external database).</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.partial-cache.max-rows</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Long</td>
+ <td>The max number of rows of lookup cache, over this value, the oldest rows will be expired.
+ "lookup.cache" must be set to "PARTIAL" to use this option. See the following <a href="#lookup-cache">Lookup Cache</a> section for more details.</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.partial-cache.expire-after-write</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>The max time to live for each rows in lookup cache after writing into the cache.
+ "lookup.cache" must be set to "PARTIAL" to use this option. See the following <a href="#lookup-cache">Lookup Cache</a> section for more details. </td>
+ </tr>
+ <tr>
+ <td><h5>lookup.partial-cache.expire-after-access</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>The max time to live for each rows in lookup cache after accessing the entry in the cache.
+ "lookup.cache" must be set to "PARTIAL" to use this option. See the following <a href="#lookup-cache">Lookup Cache</a> section for more details. </td>
+ </tr>
+ <tr>
+ <td><h5>lookup.partial-cache.caching-missing-key</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to store an empty value into the cache if the lookup key doesn't match any rows in the table.
+ "lookup.cache" must be set to "PARTIAL" to use this option.</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.max-retries</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">3</td>
+ <td>Integer</td>
+ <td>The max retry times if lookup database failed.</td>
+ </tr>
+ <tr>
+ <td><h5>lookup.retry.interval</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">1s</td>
+ <td>Duration</td>
+ <td>Specifies the retry time interval if lookup records from database failed.</td>
+ </tr>
+ <tr>
+ <td><h5>sink.buffer-flush.max-rows</h5></td>
+ <td>optional</td>
+ <td>yes</td>
+ <td style="word-wrap: break-word;">1000</td>
+ <td>Integer</td>
+ <td>Specifies the maximum number of buffered rows per batch request.</td>
+ </tr>
+ <tr>
+ <td><h5>sink.buffer-flush.interval</h5></td>
+ <td>optional</td>
+ <td>yes</td>
+ <td style="word-wrap: break-word;">1s</td>
+ <td>Duration</td>
+ <td>Specifies the batch flush interval.</td>
+ </tr>
+ <tr>
+ <td><h5>sink.max-retries</h5></td>
+ <td>optional</td>
+ <td>yes</td>
+ <td style="word-wrap: break-word;">3</td>
+ <td>Integer</td>
+ <td>The max retry times if writing records to database failed.</td>
+ </tr>
+ <tr>
+ <td><h5>sink.retry.interval</h5></td>
+ <td>optional</td>
+ <td>yes</td>
+ <td style="word-wrap: break-word;">1s</td>
+ <td>Duration</td>
+ <td>Specifies the retry time interval if writing records to database failed.</td>
+ </tr>
+ <tr>
+ <td><h5>sink.parallelism</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>Defines the parallelism of the MongoDB sink operator. By default, the parallelism is determined by the framework using the same parallelism of the upstream chained operator.</td>
+ </tr>
+ <tr>
+ <td><h5>sink.delivery-guarantee</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">at-lease-once</td>
+ <td><p>Enum</p>Possible values: none, at-least-once</td>
+ <td>Optional delivery guarantee when committing. The exactly-once guarantee is not supported yet.</td>
+ </tr>
+ </tbody>
+</table>
+
+Features
+--------
+
+### Key handling
+
+The MongoDB sink can work in either upsert mode or append mode, depending on whether a primary key
+is defined. If a primary key is defined, the MongoDB sink works in upsert mode which can consume
+queries containing UPDATE/DELETE messages. If a primary key is not defined, the MongoDB sink works
+in append mode which can only consume queries containing INSERT only messages.
+
+In MongoDB the primary key is used to calculate the MongoDB document _id.
+Its value must be unique and immutable in the collection, and may be of any
+[BSON Type](https://www.mongodb.com/docs/manual/reference/bson-types/#bson-types) other than an Array.
+If the _id contains subfields, the subfield names cannot begin with a ($) symbol.
+
+There are also some constraints on the primary key index.
+Before MongoDB 4.2, the total size of an index entry, which can include structural overhead
+depending on the BSON type, must be less than 1024 bytes.
+Starting in version 4.2, MongoDB removes the Index Key Limit.
+For more detailed introduction, you can refer to [Index Key Limit](https://www.mongodb.com/docs/manual/reference/limits/#mongodb-limit-Index-Key-Limit).
+
+The MongoDB connector generates a document _id for every row by compositing all primary key
+fields in the order defined in the DDL.
+
+- When there's only a single field in the specified primary key, we convert the field data to bson
+ value as _id of the corresponding document.
+- When there's multiple fields in the specified primary key, we convert and composite these fields
+ into a bson document as the _id of the corresponding document.
+ For example, if have a primary key statement `PRIMARY KEY (f1, f2) NOT ENFORCED`,
+ the extracted _id will be the form like `_id: {f1: v1, f2: v2}`.
+
+Notice that it will be ambiguous if the _id field exists in DDL, but the primary key is not declared as _id.
+Either use the _id column as the key, or rename the _id column.
+
+See [CREATE TABLE DDL]({{< ref "docs/dev/table/sql/create" >}}#create-table) for more details about PRIMARY KEY syntax.
+
+### Partitioned Scan
+
+To accelerate reading data in parallel `Source` task instances, Flink provides partitioned scan
+feature for MongoDB collection. The following partition strategies are provided:
+
+- `single`: treats the entire collection as a single partition.
+- `sample`: samples the collection and generate partitions which is fast but possibly uneven.
+- `split-vector`: uses the splitVector command to generate partitions for non-sharded
+ collections which is fast and even. The splitVector permission is required.
+- `sharded`: reads `config.chunks` (MongoDB splits a sharded collection into chunks, and the
+ range of the chunks are stored within the collection) as the partitions directly. The
+ sharded strategy only used for sharded collection which is fast and even. Read permission
+ of config database is required.
+- `default`: uses sharded strategy for sharded collections otherwise using split vector
+ strategy.
+
+### Lookup Cache
+
+MongoDB connector can be used in temporal join as a lookup source (aka. dimension table).
+Currently, only sync lookup mode is supported.
+
+By default, lookup cache is not enabled. You can enable it by setting `lookup.cache` to `PARTIAL`.
+
+The lookup cache is used to improve performance of temporal join the MongoDB connector.
+By default, lookup cache is not enabled, so all the requests are sent to external database.
+When lookup cache is enabled, each process (i.e. TaskManager) will hold a cache.
+Flink will lookup the cache first, and only send requests to external database when cache missing,
+and update cache with the rows returned.
+The oldest rows in cache will be expired when the cache hit to the max cached rows
+`lookup.partial-cache.max-rows` or when the row exceeds the max time to live specified by
+`lookup.partial-cache.expire-after-write` or `lookup.partial-cache.expire-after-access`.
+The cached rows might not be the latest, users can tune expiration options to a smaller value to
+have a better fresh data, but this may increase the number of requests send to database.
+So this is a balance between throughput and correctness.
+
+By default, flink will cache the empty query result for a Primary key, you can toggle the behaviour
+by setting `lookup.partial-cache.caching-missing-key` to false.
+
+### Idempotent Writes
+
+MongoDB sink will use upsert semantics rather than plain INSERT statements if primary key is defined
+in DDL. We composite the primary key fields as the document _id which is the reserved primary key of
+MongoDB. Use upsert mode to write rows into MongoDB, which provides idempotence.
+
+If there are failures, the Flink job will recover and re-process from last successful checkpoint,
+which can lead to re-processing messages during recovery. The upsert mode is highly recommended as
+it helps avoid constraint violations or duplicate data if records need to be re-processed.
+
+Data Type Mapping
+----------------
+The field data type mappings from MongoDB BSON types to Flink SQL data types are listed in the following table.
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left"><a href="https://www.mongodb.com/docs/manual/reference/bson-types/#bson-types">MongoDB BSON type</a></th>
+ <th class="text-left"><a href="{{< ref "docs/dev/table/types" >}}">Flink SQL type</a></th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><code>ObjectId</code></td>
+ <td><code>STRING</code></td>
+ </tr>
+ <tr>
+ <td><code>String</code></td>
+ <td><code>STRING</code></td>
+ </tr>
+ <tr>
+ <td><code>Boolean</code></td>
+ <td><code>BOOLEAN</code></td>
+ </tr>
+ <tr>
+ <td><code>Binary</code></td>
+ <td><code>BINARY</code><br>
+ <code>VARBINARY</code></td>
+ </tr>
+ <tr>
+ <td><code>Int32</code></td>
+ <td><code>INTEGER</code></td>
+ </tr>
+ <tr>
+ <td><code>-</code></td>
+ <td><code>TINYINT</code><br>
+ <code>SMALLINT</code><br>
+ <code>FLOAT</code></td>
+ </tr>
+ <tr>
+ <td><code>Int64</code></td>
+ <td><code>BIGINT</code></td>
+ </tr>
+ <tr>
+ <td><code>Double</code></td>
+ <td><code>DOUBLE</code></td>
+ </tr>
+ <tr>
+ <td><code>Decimal128</code></td>
+ <td><code>DECIMAL</code></td>
+ </tr>
+ <tr>
+ <td><code>DateTime</code></td>
+ <td><code>TIMESTAMP_LTZ(3)</code></td>
+ </tr>
+ <tr>
+ <td><code>Timestamp</code></td>
+ <td><code>TIMESTAMP_LTZ(0)</code></td>
+ </tr>
+ <tr>
+ <td><code>Object</code></td>
+ <td><code>ROW</code></td>
+ </tr>
+ <tr>
+ <td><code>Array</code></td>
+ <td><code>ARRAY</code></td>
+ </tr>
+ </tbody>
+</table>
+
+For specific types in MongoDB, we use [Extended JSON format](https://www.mongodb.com/docs/drivers/java/sync/current/fundamentals/data-formats/document-data-format-extended-json/)
+to map them to Flink SQL STRING type.
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left"><a href="https://www.mongodb.com/docs/manual/reference/bson-types/#bson-types">MongoDB BSON type</a></th>
+ <th class="text-left">Flink SQL STRING</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><code>Symbol</code></td>
+ <td><code>{"_value": {"$symbol": "12"}}</code></td>
+ </tr>
+ <tr>
+ <td><code>RegularExpression</code></td>
+ <td><code>{"_value": {"$regularExpression": {"pattern": "^9$", "options": "i"}}}</code></td>
+ </tr>
+ <tr>
+ <td><code>JavaScript</code></td>
+ <td><code>{"_value": {"$code": "function() { return 10; }"}}</code></td>
+ </tr>
+ <tr>
+ <td><code>DbPointer</code></td>
+ <td><code>{"_value": {"$dbPointer": {"$ref": "db.coll", "$id": {"$oid": "63932a00da01604af329e33c"}}}}</code></td>
+ </tr>
+ </tbody>
+</table>
+
+{{< top >}}
diff --git a/docs/data/mongodb.yml b/docs/data/mongodb.yml
new file mode 100644
index 0000000..23a60c2
--- /dev/null
+++ b/docs/data/mongodb.yml
@@ -0,0 +1,3 @@
+variants:
+ - maven: flink-connector-mongodb
+ sql_url: https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-mongodb/$full_version/flink-sql-connector-mongodb-$full_version.jar
diff --git a/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/MongoSourceBuilder.java b/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/MongoSourceBuilder.java
index 6ad71a7..c5252eb 100644
--- a/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/MongoSourceBuilder.java
+++ b/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/MongoSourceBuilder.java
@@ -69,7 +69,7 @@ public class MongoSourceBuilder<OUT> {
/**
* Sets the database to sink of MongoDB.
*
- * @param database the database to sink of MongoDB.
+ * @param database the database to read from MongoDB.
* @return this builder
*/
public MongoSourceBuilder<OUT> setDatabase(String database) {
@@ -80,7 +80,7 @@ public class MongoSourceBuilder<OUT> {
/**
* Sets the collection to sink of MongoDB.
*
- * @param collection the collection to sink of MongoDB.
+ * @param collection the collection to read from MongoDB.
* @return this builder
*/
public MongoSourceBuilder<OUT> setCollection(String collection) {
diff --git a/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoConnectorOptions.java b/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoConnectorOptions.java
index bda310b..a91c489 100644
--- a/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoConnectorOptions.java
+++ b/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoConnectorOptions.java
@@ -94,7 +94,7 @@ public class MongoConnectorOptions {
.intType()
.defaultValue(10)
.withDescription(
- "Specifies the the samples count per partition. It only takes effect when the partition strategy is sample. "
+ "Specifies the samples count per partition. It only takes effect when the partition strategy is sample. "
+ "The sample partitioner samples the collection, projects and sorts by the partition fields. "
+ "Then uses every 'scan.partition.samples' as the value to use to calculate the partition boundaries."
+ "The total number of samples taken is calculated as: samples per partition * (count of documents / number of documents per partition.");