This is an automated email from the ASF dual-hosted git repository.
zhangshenghang pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 73c25f0be0 [Docs][Connector-V2] Improve MongoDB connector docs (#11776)
73c25f0be0 is described below
commit 73c25f0be00a4a32c346849c5bc8e7bd660e54e4
Author: Daniel Carter <[email protected]>
AuthorDate: Wed Aug 12 21:18:37 2026 +0800
[Docs][Connector-V2] Improve MongoDB connector docs (#11776)
Co-authored-by: DanielCarter-stack
<[email protected]>
---
docs/en/connectors/sink/MongoDB.md | 168 ++++++++++-------
docs/en/connectors/source/MongoDB.md | 310 +++++++++++--------------------
docs/zh/connectors/sink/MongoDB.md | 136 ++++++++------
docs/zh/connectors/source/MongoDB.md | 351 +++++++++++++----------------------
4 files changed, 417 insertions(+), 548 deletions(-)
diff --git a/docs/en/connectors/sink/MongoDB.md
b/docs/en/connectors/sink/MongoDB.md
index 5e186c2055..e670d51e27 100644
--- a/docs/en/connectors/sink/MongoDB.md
+++ b/docs/en/connectors/sink/MongoDB.md
@@ -10,7 +10,7 @@ import ChangeLog from '../changelog/connector-mongodb.md';
> Flink<br/>
> SeaTunnel Zeta<br/>
-## Key features
+## Key Features
- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
- [x] [timer flush](../../introduction/concepts/connector-v2-features.md)
(Zeta engine only)
@@ -19,27 +19,39 @@ import ChangeLog from '../changelog/connector-mongodb.md';
**Tips**
-> 1.If you want to use CDC-written features, recommend enable the
upsert-enable configuration.
+> 1. If you want to use CDC-written features, enable the `upsert-enable`
configuration.
+> 2. Enabling `transaction` is incompatible with the Zeta timer-flush feature.
Pick one model per
+> job; mixing them will silently disable the timer.
## Description
-The MongoDB Connector provides the ability to read and write data from and to
MongoDB.
-This document describes how to set up the MongoDB connector to run data
writers against MongoDB.
+The MongoDB sink connector writes SeaTunnel rows into a MongoDB collection.
Each row is converted
+to a BSON document and sent to the configured `database` and `collection`.
+
+The connector supports two write semantics:
+
+- **Append writes** — every row produces a new document. Fast, but not
idempotent across retries.
+- **Upsert writes** — when `upsert-enable = true` and `primary-key` is
configured, the connector
+ uses the primary key as the MongoDB `_id` (or compound `_id`) and upserts.
Combined with
+ checkpoint-based recovery this gives at-least-once with idempotent retries,
which is the standard
+ way to deliver exactly-once into MongoDB.
+
+Buffering, retry, and the optional transaction are tuned via the options below.
## Supported DataSource Info
-In order to use the Mongodb connector, the following dependencies are required.
-They can be downloaded via install-plugin.sh or from the Maven central
repository.
+In order to use the MongoDB connector, the following dependency is required.
+It can be downloaded via `install-plugin.sh` or from the Maven central
repository.
| Datasource | Supported Versions | Dependency
|
|------------|--------------------|---------------------------------------------------------------------------------------|
-| MongoDB | universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-mongodb)
|
+| MongoDB | Universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-mongodb)
|
## Data Type Mapping
-The following table lists the field data type mapping from MongoDB BSON type
to Seatunnel data type.
+The following table lists the field data type mapping from SeaTunnel data type
to MongoDB BSON type.
-| Seatunnel Data Type | MongoDB BSON Type |
+| SeaTunnel Data Type | MongoDB BSON Type |
|---------------------|-------------------|
| STRING | ObjectId |
| STRING | String |
@@ -59,38 +71,50 @@ The following table lists the field data type mapping from
MongoDB BSON type to
**Tips**
-> 1.When using SeaTunnel to write Date and Timestamp types to MongoDB, both
will produce a Date data type in MongoDB, but the precision will be different.
The data generated by the SeaTunnel Date type has second-level precision, while
the data generated by the SeaTunnel Timestamp type has millisecond-level
precision.<br/>
-> 2.When using the DECIMAL type in SeaTunnel, be aware that the maximum range
cannot exceed 34 digits, which means you should use decimal(34, 18).<br/>
+> 1. When SeaTunnel writes `Date` and `Timestamp` types to MongoDB, both
become MongoDB `Date`
+> fields, but at different precisions: SeaTunnel `Date` is second
precision; SeaTunnel
+> `Timestamp` is millisecond precision.
+> 2. When using the `DECIMAL` type in SeaTunnel, the maximum range cannot
exceed 34 digits. Use
+> `decimal(34, 18)` to stay within the supported precision and scale.
## Sink Options
-| Name | Type | Required | Default | Description
|
-|-----------------------|----------|----------|--------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| uri | String | Yes | - | The MongoDB standard
connection uri. eg.
mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true.
|
-| database | String | Yes | - | The name of the
MongoDB database to read or write to. When configuring multiple tables at the
source, you can use `${database_name}` as a placeholder, for example: `database
= "${database_name}_test_database"` .
|
-| collection | String | Yes | - | The name of the
MongoDB collection to read or write. When configuring multiple tables at the
source end, you can use `${database_name}`, `${schema_name}`, and
`${table_name}` as placeholders, for example: `collection =
"${database_name}_${schema_name}_${table_name}_check"` |
-| buffer-flush.max-rows | Int | No | 1000 | Specifies the maximum
number of buffered rows per batch request.
|
-| buffer-flush.interval | Long | No | 30000 | Specifies the maximum
interval of buffered rows per batch request, the unit is millisecond.
|
-| retry.max | Int | No | 3 | Specifies the max
number of retry if writing records to database failed.
|
-| retry.interval | Long | No | 1000 | Specifies the retry
time interval if writing records to database failed, the unit is millisecond.
|
-| upsert-enable | Boolean | No | false | Whether to write
documents via upsert mode.
|
-| primary-key | List | No | - | The primary keys for
upsert/update. Keys are in `["id","name",...]` format for properties.
|
-| transaction | Boolean | No | false | Whether to use
transactions in MongoSink (requires MongoDB 4.2+).
|
-| common-options | | No | - | Sink plugin common
parameters, please refer to [Sink Common
Options](../common-options/sink-common-options.md) for details
|
-| data_save_mode | Enum | No | APPEND_DATA | The data
saving mode of mongodb. Supported values: `DROP_DATA`, `APPEND_DATA`, and
`ERROR_WHEN_DATA_EXISTS`. |
-
+| Name | Type | Required | Default Value | Description
|
+|-----------------------|----------|----------|---------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| uri | String | Yes | - | The MongoDB
standard connection URI, for example
`mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true`.
See [Parameter Interpretation](#parameter-interpretation) for more URI
samples. |
+| database | String | Yes | - | The name of
the MongoDB database to write to. When writing multiple tables from the source,
you can use `${database_name}` as a placeholder, for example `database =
"${database_name}_test_database"`.
|
+| collection | String | Yes | - | The name of
the MongoDB collection to write to. When writing multiple tables from the
source, you can use `${database_name}`, `${schema_name}`, and `${table_name}`
as placeholders, for example `collection =
"${database_name}_${schema_name}_${table_name}_check"`. |
+| buffer-flush.max-rows | Int | No | 1000 | The maximum
number of buffered rows per batch request.
|
+| buffer-flush.interval | Long | No | 30000 | The maximum
interval (in milliseconds) of buffered rows per batch request.
|
+| retry.max | Int | No | 3 | The maximum
number of retries if writing records to MongoDB fails.
|
+| retry.interval | Long | No | 1000 | The retry
interval (in milliseconds) if writing records to MongoDB fails.
|
+| upsert-enable | Boolean | No | false | Whether to
write documents via upsert mode. When enabled, `primary-key` must also be
configured.
|
+| primary-key | List | No | - | The primary
keys used for upsert/update. The list format is `["id","name",...]`.
|
+| transaction | Boolean | No | false | Whether to use
transactions in MongoSink (requires MongoDB 4.2+).
|
+| data_save_mode | Enum | No | APPEND_DATA | The data
saving mode for the MongoDB collection. Supported values: `DROP_DATA` (truncate
the collection before writing), `APPEND_DATA` (append to existing data),
`ERROR_WHEN_DATA_EXISTS` (fail if the collection already has data).
|
+| common-options | | No | - | Sink plugin
common parameters, please refer to [Sink Common
Options](../common-options/sink-common-options.md) for details.
|
### Tips
-> 1.The connector-level data flushing logic of MongoDB Sink is jointly
controlled by three parameters: `buffer-flush.max-rows`,
`buffer-flush.interval`, and `checkpoint.interval`.<br/>
-> Data flushing will be triggered if any of these conditions are met.<br/>
-> 2.Compatible with the historical parameter `upsert-key`. If `upsert-key` is
set, please do not set `primary-key`.<br/>
+> 1. The connector-level data flushing logic is jointly controlled by three
parameters:
+> `buffer-flush.max-rows`, `buffer-flush.interval`, and
`checkpoint.interval`. Whichever is
+> reached first triggers the flush.
+> 2. The legacy option name `upsert-key` is still accepted as a fallback for
`primary-key`. Do not
+> set both at the same time.
+> 3. The `transaction` option is incompatible with the Zeta timer-flush
feature described below.
+> Enable exactly one model per job.
### Zeta Timer Flush
-This engine-level feature is supported only by Zeta. Spark and Flink do not
inject `FlushSignal` records. On Zeta, configure `sink.flush.interval` in the
`env` block to flush pending bulk requests even when `buffer-flush.max-rows`
has not been reached. Unlike `buffer-flush.interval`, the engine timer does not
require a new input record to trigger the check.
+This engine-level feature is supported only by Zeta. Spark and Flink do not
inject `FlushSignal`
+records. On Zeta, configure `sink.flush.interval` in the `env` block to flush
pending bulk requests
+even when `buffer-flush.max-rows` has not been reached. Unlike
`buffer-flush.interval`, the engine
+timer does not require a new input record to trigger the check.
-Timer flush is enabled only when `transaction = false`. MongoDB transaction
mode is committed through checkpoints, so timer flush is disabled to preserve
the transaction boundary. The initial timer-flush implementation provides
at-least-once delivery rather than 2PC exactly-once. Enabling upsert with
deterministic primary keys can make retries idempotent.
+Timer flush is enabled only when `transaction = false`. MongoDB transaction
mode is committed
+through checkpoints, so timer flush is disabled to preserve the transaction
boundary. The initial
+timer-flush implementation provides at-least-once delivery rather than 2PC
exactly-once. Enabling
+upsert with deterministic primary keys can make retries idempotent.
```hocon
env {
@@ -110,35 +134,34 @@ sink {
}
```
-## How to Create a MongoDB Data Synchronization Jobs
+## How to Create a MongoDB Data Synchronization Job
-The following example demonstrates how to create a data synchronization job
that writes randomly generated data to a MongoDB database:
+The following example writes randomly generated data into a MongoDB collection:
-```bash
-# Set the basic configuration of the task to be performed
+```hocon
env {
parallelism = 1
job.mode = "BATCH"
- checkpoint.interval = 1000
+ checkpoint.interval = 1000
}
source {
FakeSource {
- row.num = 2
- bigint.min = 0
- bigint.max = 10000000
- split.num = 1
- split.read-interval = 300
- schema {
- fields {
- c_bigint = bigint
- }
+ row.num = 2
+ bigint.min = 0
+ bigint.max = 10000000
+ split.num = 1
+ split.read-interval = 300
+ schema {
+ fields {
+ c_bigint = bigint
}
}
+ }
}
sink {
- MongoDB{
+ MongoDB {
uri = "mongodb://user:[email protected]:27017"
database = "test"
collection = "test"
@@ -148,8 +171,8 @@ sink {
### Multiple Table Write
-When upstream records carry table metadata, `database` and `collection` can
use placeholders. The common placeholders are
-`${database_name}`, `${schema_name}`, and `${table_name}`.
+When upstream records carry table metadata, `database` and `collection` can
use placeholders. The
+common placeholders are `${database_name}`, `${schema_name}`, and
`${table_name}`.
```hocon
source {
@@ -191,7 +214,7 @@ source {
sink {
MongoDB {
- uri = "mongodb://user:[email protected]:27017/test_db?retryWrites=true"
+ uri = "mongodb://127.0.0.1:27017/test_db?retryWrites=true"
database = "test_db"
collection = "${database_name}_${schema_name}_${table_name}_check"
}
@@ -205,44 +228,45 @@ sink {
Unauthenticated single node connection:
```bash
-mongodb://127.0.0.0:27017/mydb
+mongodb://127.0.0.1:27017/mydb
```
Replica set connection:
```bash
-mongodb://127.0.0.0:27017/mydb?replicaSet=xxx
+mongodb://127.0.0.1:27017/mydb?replicaSet=xxx
```
Authenticated replica set connection:
```bash
-mongodb://admin:[email protected]:27017/mydb?replicaSet=xxx&authSource=admin
+mongodb://admin:[email protected]:27017/mydb?replicaSet=xxx&authSource=admin
```
Multi-node replica set connection:
```bash
-mongodb://127.0.0..1:27017,127.0.0..2:27017,127.0.0.3:27017/mydb?replicaSet=xxx
+mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb?replicaSet=xxx
```
-Sharded cluster connection:
+Sharded cluster connection (route through one `mongos`):
```bash
-mongodb://127.0.0.0:27017/mydb
+mongodb://mongos1.example.com:27017,mongos2.example.com:27017,mongos3.example.com:27017/mydb
```
-Multiple mongos connections:
+Multiple mongos connections (comma-separated list of mongos hosts):
```bash
mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb
```
-Note: The username and password in the URI must be URL-encoded before being
concatenated into the connection string.
+> Note: The username and password in the URI must be URL-encoded before being
concatenated into
+> the connection string.
### Buffer Flush
-```bash
+```hocon
sink {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
@@ -254,32 +278,38 @@ sink {
}
```
-### Why is Not Recommended to Use Transactions for Operation?
+### Why Is It Not Recommended to Use Transactions for Every Operation?
-Although MongoDB has fully supported multi-document transactions since version
4.2, it doesn't mean that everyone should use them recklessly.
-Transactions are equivalent to locks, node coordination, additional overhead,
and performance impact.
-Instead, the principle for using transactions should be: avoid using them if
possible.
-The necessity for using transactions can be greatly avoided by designing
systems rationally.
+Although MongoDB has fully supported multi-document transactions since version
4.2, this does not
+mean that every workload should use them. Transactions bring locking, node
coordination, extra
+round trips, and performance overhead. The guiding principle is: avoid
transactions whenever
+possible. A well-designed pipeline can usually make idempotent writes enough.
### Idempotent Writes
-By specifying a clear primary key and using the upsert method, exactly-once
write semantics can be achieved.
+By specifying a clear primary key and using the upsert method, exactly-once
write semantics can be
+achieved.
-If `primary-key` and `upsert-enable` is defined in the configuration, the
MongoDB sink will use upsert semantics instead of regular INSERT statements. We
combine the primary keys declared in upsert-key as the MongoDB reserved primary
key and use upsert mode for writing to ensure idempotent writes.
-In the event of a failure, Seatunnel jobs will recover from the last
successful checkpoint and reprocess, which may result in duplicate message
processing during recovery. It is highly recommended to use upsert mode, as it
helps to avoid violating database primary key constraints and generating
duplicate data if records need to be reprocessed.
+If `primary-key` and `upsert-enable` are defined in the configuration, the
MongoDB sink uses
+upsert semantics instead of regular INSERT statements. The connector combines
the primary keys
+declared in `primary-key` as the MongoDB reserved primary key and writes via
upsert mode to ensure
+idempotent writes. In the event of a failure, SeaTunnel jobs recover from the
last successful
+checkpoint and reprocess, which may result in duplicate processing during
recovery. It is highly
+recommended to use upsert mode because it avoids violating database primary
key constraints and
+generating duplicate data if records need to be reprocessed.
-```bash
+```hocon
sink {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
database = "test_db"
collection = "users"
upsert-enable = true
- primary-key = ["name","status"]
+ primary-key = ["name", "status"]
}
}
```
## Changelog
-<ChangeLog />
+<ChangeLog />
\ No newline at end of file
diff --git a/docs/en/connectors/source/MongoDB.md
b/docs/en/connectors/source/MongoDB.md
index 0cfb58afef..92a7adc6bb 100644
--- a/docs/en/connectors/source/MongoDB.md
+++ b/docs/en/connectors/source/MongoDB.md
@@ -13,7 +13,7 @@ import ChangeLog from '../changelog/connector-mongodb.md';
## Key Features
- [x] [batch](../../introduction/concepts/connector-v2-features.md)
-- [ ] [stream](../../introduction/concepts/connector-v2-features.md)
+- [x] [stream](../../introduction/concepts/connector-v2-features.md)
- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
- [x] [column projection](../../introduction/concepts/connector-v2-features.md)
- [x] [parallelism](../../introduction/concepts/connector-v2-features.md)
@@ -21,23 +21,34 @@ import ChangeLog from '../changelog/connector-mongodb.md';
## Description
-The MongoDB Connector provides the ability to read and write data from and to
MongoDB.
-This document describes how to set up the MongoDB connector to run data reads
against MongoDB.
+The MongoDB source connector reads documents from a MongoDB collection and
converts each BSON
+document into a SeaTunnel row. It supports both **batch** and **streaming**
jobs, and reads in
+parallel by splitting the source collection across `partition.split-key`
ranges.
+
+You can narrow what is read and what columns are returned without scanning the
whole collection:
+
+- Use `match.query` to filter documents by a MongoDB query expression.
+- Use `match.projection` to control which fields appear in the result.
+- Use `flat.sync-string` to capture the whole document as one JSON `STRING`
column when no
+ fixed schema is needed.
+
+In streaming mode the connector reads the assigned splits and tracks progress
through checkpoints
+so a restarted job resumes from the last committed cursor.
## Supported DataSource Info
-In order to use the Mongodb connector, the following dependencies are required.
-They can be downloaded via install-plugin.sh or from the Maven central
repository.
+In order to use the MongoDB connector, the following dependency is required.
+It can be downloaded via `install-plugin.sh` or from the Maven central
repository.
| Datasource | Supported Versions | Dependency
|
|------------|--------------------|---------------------------------------------------------------------------------------|
-| MongoDB | universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-mongodb)
|
+| MongoDB | Universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-mongodb)
|
## Data Type Mapping
The following table lists the field data type mapping from MongoDB BSON type
to SeaTunnel data type.
-| MongoDB BSON type | SeaTunnel Data type |
+| MongoDB BSON type | SeaTunnel Data Type |
|-------------------|---------------------|
| ObjectId | STRING |
| String | STRING |
@@ -52,7 +63,8 @@ The following table lists the field data type mapping from
MongoDB BSON type to
| Object | ROW |
| Array | ARRAY |
-For specific types in MongoDB, we use Extended JSON format to map them to
SeaTunnel STRING type.
+For specific types in MongoDB, the connector uses Extended JSON format and
maps them to the
+SeaTunnel `STRING` type.
| MongoDB BSON type | SeaTunnel STRING
|
|-------------------|----------------------------------------------------------------------------------------------|
@@ -63,42 +75,45 @@ For specific types in MongoDB, we use Extended JSON format
to map them to SeaTun
**Tips**
-> 1.When using the DECIMAL type in SeaTunnel, be aware that the maximum range
cannot exceed 34 digits, which means you should use decimal(34, 18).<br/>
+> 1. When using the `DECIMAL` type in SeaTunnel, the maximum range cannot
exceed 34 digits. Use
+> `decimal(34, 18)` to stay within the supported precision and scale.
## Source Options
-| Name | Type | Required | Default |
Description
|
-|----------------------|---------|----------|------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| uri | String | Yes | - | The MongoDB
standard connection uri. eg.
mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true.
|
-| database | String | Yes | - | The name of
MongoDB database to read or write.
|
-| collection | String | Yes | - | The name of
MongoDB collection to read or write.
|
-| schema | Config | Yes | - | MongoDB's
BSON and SeaTunnel data structure mapping. For more details, please refer to
[Schema Feature](../../introduction/concepts/schema-feature.md).
|
-| match.query | String | No | - | In MongoDB,
filters are used to filter documents for query operations.
|
-| match.projection | String | No | - | In MongoDB,
Projection is used to control the fields contained in the query results.
|
-| partition.split-key | String | No | _id | The key of
Mongodb fragmentation.
|
-| partition.split-size | Long | No | 64 * 1024 * 1024 | The size of
Mongodb fragment.
|
-| cursor.no-timeout | Boolean | No | true | 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. |
-| fetch.size | Int | No | 2048 | Set the
number of documents obtained from the server for each batch. Setting the
appropriate batch size can improve query performance and avoid the memory
pressure caused by obtaining a large amount of data at one time.
|
-| max.time-min | Long | No | 10 | This
parameter is a MongoDB query option that limits the maximum execution time for
query operations. The value of maxTimeMin is in minutes. If the execution time
of the query exceeds the specified time limit, MongoDB will terminate the
operation and return an error. |
-| flat.sync-string | Boolean | No | false | When enabled,
the connector maps the whole MongoDB document to one SeaTunnel `STRING` field.
The schema must contain exactly one field and that field must be `STRING`.
|
-| common-options | | No | - | Source plugin
common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details
|
+| Name | Type | Required | Default |
Description
|
+|-----------------------|---------|----------|--------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| uri | String | Yes | - | The
MongoDB standard connection URI, for example
`mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true`.
See [Parameter Interpretation](#parameter-interpretation) for more URI
samples. |
+| database | String | Yes | - | The name
of the MongoDB database to read from.
|
+| collection | String | Yes | - | The name
of the MongoDB collection to read from.
|
+| schema | Config | Yes | - | The
mapping between MongoDB's BSON and SeaTunnel data structure. For more details,
see [Schema Feature](../../introduction/concepts/schema-feature.md).
|
+| match.query | String | No | - | MongoDB
query expression used to filter documents for read operations. Compatible with
the legacy option name `matchQuery`.
|
+| match.projection | String | No | - | MongoDB
projection expression used to control which fields appear in the result.
|
+| partition.split-key | String | No | _id | The field
used as the MongoDB split key. The connector splits the collection by the value
range of this key.
|
+| partition.split-size | Long | No | 64 * 1024 * 1024 | The size
of each MongoDB split. Smaller split sizes increase the number of splits and
parallelism, while larger sizes reduce it.
|
+| cursor.no-timeout | Boolean | No | true | MongoDB
server normally times out idle cursors after 10 minutes of inactivity to
reclaim memory. Set this option to `true` to keep the cursor open across
long-running batches. If the application holds the batch for more than 30
minutes, MongoDB marks the session as expired and closes it. |
+| fetch.size | Int | No | 2048 | The number
of documents obtained from the server per batch. Tuning this value balances
query performance and memory pressure.
|
+| max.time-min | Long | No | 10 | The
maximum execution time (in minutes) for each MongoDB query. MongoDB terminates
the operation and returns an error when this limit is exceeded.
|
+| flat.sync-string | Boolean | No | false | When
enabled, the connector maps the whole MongoDB document into one SeaTunnel
`STRING` field. The schema must contain exactly one field and that field must
be of type `STRING`.
|
+| common-options | | No | - | Source
plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details.
|
### Tips
-> 1.The parameter `match.query` is compatible with the historical old version
parameter `matchQuery`, and they are equivalent replacements.<br/>
+> 1. The `match.query` option is compatible with the legacy option name
`matchQuery`; they are equivalent.
+> 2. Use `partition.split-key` together with `partition.split-size` to control
parallel reads. The
+> split key should reference an indexed field for best performance.
+> 3. When `flat.sync-string = true`, the configured schema is ignored except
for the single `STRING`
+> field that receives the document.
-## How to Create a MongoDB Data Synchronization Jobs
+## How to Create a MongoDB Data Synchronization Job
-The following example demonstrates how to create a data synchronization job
that reads data from MongoDB and prints it on the local client:
+The following example reads data from MongoDB and prints it on the local
client:
-```bash
-# Set the basic configuration of the task to be performed
+```hocon
env {
parallelism = 1
job.mode = "BATCH"
}
-# Create a source to connect to Mongodb
source {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
@@ -135,7 +150,6 @@ source {
}
}
-# Console printing of the read Mongodb data
sink {
Console {
parallelism = 1
@@ -171,26 +185,28 @@ Multi-node replica set connection:
mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb?replicaSet=xxx
```
-Sharded cluster connection:
+Sharded cluster connection (route through one `mongos`):
```bash
-mongodb://192.168.0.100:27017/mydb
+mongodb://mongos1.example.com:27017,mongos2.example.com:27017,mongos3.example.com:27017/mydb
```
-Multiple mongos connections:
+Multiple mongos connections (comma-separated list of mongos hosts):
```bash
mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb
```
-Note: The username and password in the URI must be URL-encoded before being
concatenated into the connection string.
+> Note: The username and password in the URI must be URL-encoded before being
concatenated into
+> the connection string.
### MatchQuery Scan
-In data synchronization scenarios, the matchQuery approach needs to be used
early to reduce the number of documents that need to be processed by subsequent
operators, thus improving performance.
-Here is a simple example of a seatunnel using `match.query`
+In data synchronization scenarios, the `match.query` approach should be used
early to reduce the
+number of documents that need to be processed by downstream operators,
improving overall
+performance. Here is a simple example of using `match.query`:
-```bash
+```hocon
source {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
@@ -207,38 +223,41 @@ source {
}
```
-The following are examples of MatchQuery query statements of various data
types:
+The following are examples of `match.query` expressions for various data types:
```bash
# Query Boolean type
-"{c_boolean:true}"
+"{c_boolean: true}"
# Query string type
-"{c_string:\"OCzCj\"}"
+"{c_string: \"OCzCj\"}"
# Query the integer
-"{c_int:2}"
-# Type of query time
-"{c_date:ISODate(\"2023-06-26T16:00:00.000Z\")}"
-# Query floating point type
-{c_double:{$gte:1.71763202185342e+308}}
+"{c_int: 2}"
+# Query the date type
+"{c_date: {\$date: \"2023-06-26T16:00:00.000Z\"}}"
+# Query the floating point type
+"{c_double: {\$gte: 1.71763202185342e+308}}"
```
-Please refer to how to write the syntax of
`match.query`：https://www.mongodb.com/docs/manual/tutorial/query-documents
+Refer to the MongoDB manual for the full query syntax:
+<https://www.mongodb.com/docs/manual/tutorial/query-documents>
### Projection Scan
-In MongoDB, Projection is used to control which fields are included in the
query results. This can be accomplished by specifying which fields need to be
returned and which fields do not.
-In the find() method, a projection object can be passed as a second argument.
The key of the projection object indicates the fields to include or exclude,
and a value of 1 indicates inclusion and 0 indicates exclusion.
-Here is a simple example, assuming we have a collection named users:
+In MongoDB, projection controls which fields appear in the query results by
specifying which
+fields are returned and which are excluded. In the `find()` method, a
projection object can be
+passed as the second argument. A value of `1` includes the field, `0` excludes
it. For example,
+given a `users` collection:
-```bash
-# Returns only the name and email fields
+```javascript
+// Returns only the `name` field and excludes the `email` field
db.users.find({}, { name: 1, email: 0 });
```
-In data synchronization scenarios, projection needs to be used early to reduce
the number of documents that need to be processed by subsequent operators, thus
improving performance.
-Here is a simple example of a seatunnel using projection:
+In data synchronization scenarios, projection should be used early to reduce
the number of fields
+that need to be processed by downstream operators. Here is a simple example of
using projection in
+SeaTunnel:
-```bash
+```hocon
source {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
@@ -252,15 +271,15 @@ source {
}
}
}
-
```
### Partitioned Scan
-To speed up reading data in parallel source task instances, seatunnel provides
a partitioned scan feature for MongoDB collections. The following partitioning
strategies are provided.
-Users can control data sharding by setting the partition.split-key for
sharding keys and partition.split-size for sharding size.
+To speed up reading data in parallel source tasks, SeaTunnel provides a
partitioned scan feature
+for MongoDB collections. Configure `partition.split-key` for the split field
and
+`partition.split-size` for the split size to control data sharding:
-```bash
+```hocon
source {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
@@ -276,17 +295,20 @@ source {
}
}
}
-
```
+> Tip: Pick a split key backed by an index so the source can fast-skip through
the collection when
+> enumerating split ranges.
+
### Flat Sync String
-By utilizing `flat.sync-string`, only one field attribute value can be set,
and the field type must be a String.
-This operation will perform a string mapping on a single MongoDB data entry.
+By enabling `flat.sync-string`, you only need to declare a single field whose
type is `STRING`. The
+connector will serialize each MongoDB document as an Extended JSON string and
put it into that
+field.
-```bash
+```hocon
env {
- parallelism = 10
+ parallelism = 1
job.mode = "BATCH"
}
source {
@@ -307,147 +329,41 @@ sink {
}
```
-Use the data samples synchronized with modified parameters, such as the
following:
+A sample document that flows through this configuration looks like the
following:
```json
{
- "_id":{
- "$oid":"643d41f5fdc6a52e90e59cbf"
+ "_id": {
+ "$oid": "643d41f5fdc6a52e90e59cbf"
},
- "c_map":{
- "OQBqH":"jllt",
- "rkvlO":"pbfdf",
- "pCMEX":"hczrdtve",
- "DAgdj":"t",
- "dsJag":"voo"
+ "c_map": {
+ "OQBqH": "jllt",
+ "rkvlO": "pbfdf",
+ "pCMEX": "hczrdtve",
+ "DAgdj": "t",
+ "dsJag": "voo"
},
- "c_array":[
- {
- "$numberInt":"-865590937"
- },
- {
- "$numberInt":"833905600"
- },
- {
- "$numberInt":"-1104586446"
- },
- {
- "$numberInt":"2076336780"
- },
- {
- "$numberInt":"-1028688944"
- }
+ "c_array": [
+ { "$numberInt": "-865590937" },
+ { "$numberInt": "833905600" },
+ { "$numberInt": "-1104586446" },
+ { "$numberInt": "2076336780" },
+ { "$numberInt": "-1028686444" }
],
- "c_string":"bddkzxr",
- "c_boolean":false,
- "c_tinyint":{
- "$numberInt":"39"
- },
- "c_smallint":{
- "$numberInt":"23672"
- },
- "c_int":{
- "$numberInt":"-495763561"
- },
- "c_bigint":{
- "$numberLong":"3768307617923954543"
- },
- "c_float":{
- "$numberDouble":"5.284220288280258E37"
- },
- "c_double":{
- "$numberDouble":"1.1706091642478246E308"
- },
- "c_bytes":{
- "$binary":{
- "base64":"ZWJ4",
- "subType":"00"
- }
- },
- "c_date":{
- "$date":{
- "$numberLong":"1686614400000"
- }
- },
- "c_decimal":{
- "$numberDecimal":"683265300"
- },
- "c_timestamp":{
- "$date":{
- "$numberLong":"1684283772000"
- }
- },
- "c_row":{
- "c_map":{
- "OQBqH":"cbrzhsktmm",
- "rkvlO":"qtaov",
- "pCMEX":"tuq",
- "DAgdj":"jzop",
- "dsJag":"vwqyxtt"
- },
- "c_array":[
- {
- "$numberInt":"1733526799"
- },
- {
- "$numberInt":"-971483501"
- },
- {
- "$numberInt":"-1716160960"
- },
- {
- "$numberInt":"-919976360"
- },
- {
- "$numberInt":"727499700"
- }
- ],
- "c_string":"oboislr",
- "c_boolean":true,
- "c_tinyint":{
- "$numberInt":"-66"
- },
- "c_smallint":{
- "$numberInt":"1308"
- },
- "c_int":{
- "$numberInt":"-1573886733"
- },
- "c_bigint":{
- "$numberLong":"4877994302999518682"
- },
- "c_float":{
- "$numberDouble":"1.5353209063652051E38"
- },
- "c_double":{
- "$numberDouble":"1.1952441956458565E308"
- },
- "c_bytes":{
- "$binary":{
- "base64":"cWx5Ymp0Yw==",
- "subType":"00"
- }
- },
- "c_date":{
- "$date":{
- "$numberLong":"1686614400000"
- }
- },
- "c_decimal":{
- "$numberDecimal":"656406177"
- },
- "c_timestamp":{
- "$date":{
- "$numberLong":"1684283772000"
- }
- }
- },
- "id":{
- "$numberInt":"2"
- }
+ "c_string": "bddkzxr",
+ "c_boolean": false,
+ "c_tinyint": { "$numberInt": "39" },
+ "c_smallint": { "$numberInt": "23672" },
+ "c_int": { "$numberInt": "-495763561" },
+ "c_bigint": { "$numberLong": "3768307617923954543" },
+ "c_double": { "$numberDouble": "1.1706091642478246E308" },
+ "c_bytes": { "$binary": { "base64": "ZWJ4", "subType": "00" } },
+ "c_date": { "$date": { "$numberLong": "1686614400000" } },
+ "c_decimal": { "$numberDecimal": "683265300" },
+ "c_timestamp": { "$date": { "$numberLong": "1684283772000" } }
}
```
## Changelog
-<ChangeLog />
+<ChangeLog />
\ No newline at end of file
diff --git a/docs/zh/connectors/sink/MongoDB.md
b/docs/zh/connectors/sink/MongoDB.md
index e97ccaa10e..fb0c039642 100644
--- a/docs/zh/connectors/sink/MongoDB.md
+++ b/docs/zh/connectors/sink/MongoDB.md
@@ -20,16 +20,27 @@ import ChangeLog from '../changelog/connector-mongodb.md';
**提示**
> 1. 如果希望使用 CDC 写入功能，建议启用 `upsert-enable` 配置项。
+> 2. 启用 `transaction` 与 Zeta 定时刷新互斥，请在同一个作业中选择其中一种模式；混用会导致
+> 定时刷新被静默关闭。
## 介绍
-MongoDB 连接器提供从 MongoDB 读取数据以及向 MongoDB 写入数据的能力。
-本文档将介绍如何配置 MongoDB 连接器，以便执行向 MongoDB 写入数据的任务。
+MongoDB Sink 连接器将 SeaTunnel 行记录写入 MongoDB 集合。每行记录都会被转换为 BSON 文档
+发送到所配置的 `database` 和 `collection`。
+
+连接器支持两种写入语义：
+
+- **追加写入**：每行生成一条新文档。性能最好，但在失败重试场景下不幂等。
+- **Upsert 写入**：当 `upsert-enable = true` 且配置了 `primary-key` 时，连接器会把主键作为
+ MongoDB 的 `_id`（或复合 `_id`）并以 upsert 方式写入。结合 checkpoint 恢复机制，可以实现
+ at-least-once + 幂等重试，这是把 exactly-once 落到 MongoDB 的标准做法。
+
+缓存、重试以及可选的事务都可以通过下文的配置项进行调节。
## 支持的数据源信息
-要使用 MongoDB 连接器，需要以下依赖。
-可通过 `install-plugin.sh` 下载，或从 Maven 中央仓库获取。
+要使用 MongoDB 连接器，需要以下依赖。可以通过 `install-plugin.sh` 下载，也可以从 Maven
+中央仓库获取。
| 数据源 | 支持版本 | 依赖 |
|---------|------------|---------|
@@ -37,7 +48,7 @@ MongoDB 连接器提供从 MongoDB 读取数据以及向 MongoDB 写入数据的
## 数据类型映射
-以下表格展示了 MongoDB BSON 类型与 SeaTunnel 数据类型之间的映射关系。
+下表展示了 SeaTunnel 数据类型到 MongoDB BSON 类型的映射关系。
| SeaTunnel 数据类型 | MongoDB BSON 类型 |
|--------------------|-------------------|
@@ -59,37 +70,45 @@ MongoDB 连接器提供从 MongoDB 读取数据以及向 MongoDB 写入数据的
**提示**
-> 1. 使用 SeaTunnel 将 `Date` 和 `Timestamp` 类型写入 MongoDB 时，MongoDB 中都会生成 `Date`
类型字段，但精度不同：SeaTunnel 的 `Date` 类型精度为秒，`Timestamp` 类型精度为毫秒。<br/>
-> 2. 当使用 `DECIMAL` 类型时，最大精度不能超过 34 位，也就是说应使用 `decimal(34, 18)`。
+> 1. 使用 SeaTunnel 将 `Date` 和 `Timestamp` 类型写入 MongoDB 时，结果都是 `Date` 类型，但
+> 精度不同：SeaTunnel 的 `Date` 为秒级精度，`Timestamp` 为毫秒级精度。
+> 2. 在 SeaTunnel 中使用 `DECIMAL` 类型时，最大精度不能超过 34 位。建议使用
+> `decimal(34, 18)` 以满足支持的精度与标度。
## Sink 参数说明
-| 参数名称 | 类型 | 是否必填 | 默认值 | 说明 |
-|-----------------------|----------|----------|--------|------|
-| uri | String | 是 | - | MongoDB 标准连接
URI，例如：`mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true`。
|
-| database | String | 是 | - | 要读取或写入的 MongoDB
数据库名称。配置多表同步时，可使用占位符 `${database_name}`，例如：`database =
"${database_name}_test_database"`。 |
-| collection | String | 是 | - | 要读取或写入的 MongoDB
集合名称。配置多表同步时，可使用 `${database_name}`、`${schema_name}`、`${table_name}`
等占位符，例如：`collection = "${database_name}_${schema_name}_${table_name}_check"`。 |
-| buffer-flush.max-rows | Int | 否 | 1000 | 每次批量写入请求的最大缓存行数。 |
-| buffer-flush.interval | Long | 否 | 30000 | 批量写入的最大时间间隔（毫秒）。 |
-| retry.max | Int | 否 | 3 | 写入失败时的最大重试次数。 |
-| retry.interval | Long | 否 | 1000 | 写入失败后的重试间隔时间（毫秒）。 |
-| upsert-enable | Boolean | 否 | false | 是否启用 upsert 模式进行写入。 |
-| primary-key | List | 否 | - | 用于 upsert 或更新操作的主键，格式为
`["id","name",...]`。 |
-| transaction | Boolean | 否 | false | 是否在 MongoSink 中使用事务（需要
MongoDB 4.2+）。 |
-| common-options | - | 否 | - | 通用 Sink 插件参数，详见 [Sink
Common Options](../common-options/sink-common-options.md)。 |
-| data_save_mode | Enum | 否 | APPEND_DATA |
数据写入模式：`DROP_DATA` 表示插入数据前清空集合，`APPEND_DATA` 表示追加数据，`ERROR_WHEN_DATA_EXISTS`
表示集合已有数据时报错。 |
+| 参数名称 | 类型 | 是否必填 | 默认值 | 描述 |
+|-----------------------|----------|----------|----------------|------|
+| uri | String | 是 | - | MongoDB 标准连接
URI，例如
`mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true`。更多示例请参考
[参数详解](#参数详解)。 |
+| database | String | 是 | - | 要写入的 MongoDB
数据库名称。配置多表同步时，可使用占位符 `${database_name}`，例如：`database =
"${database_name}_test_database"`。 |
+| collection | String | 是 | - | 要写入的 MongoDB
集合名称。配置多表同步时，可使用 `${database_name}`、`${schema_name}`、`${table_name}`
等占位符，例如：`collection = "${database_name}_${schema_name}_${table_name}_check"`。 |
+| buffer-flush.max-rows | Int | 否 | 1000 |
每次批量写入请求的最大缓存行数。 |
+| buffer-flush.interval | Long | 否 | 30000 |
批量写入请求的最大时间间隔（毫秒）。 |
+| retry.max | Int | 否 | 3 | 写入失败时的最大重试次数。 |
+| retry.interval | Long | 否 | 1000 |
写入失败后的重试间隔（毫秒）。 |
+| upsert-enable | Boolean | 否 | false | 是否启用 upsert
模式写入。开启时需要同时配置 `primary-key`。 |
+| primary-key | List | 否 | - | 用于 upsert
或更新的主键，格式为 `["id","name",...]`。 |
+| transaction | Boolean | 否 | false | 是否在 MongoSink
中启用事务（需要 MongoDB 4.2+）。 |
+| data_save_mode | Enum | 否 | APPEND_DATA | MongoDB
集合的数据写入模式：`DROP_DATA` 表示写入前清空集合；`APPEND_DATA` 表示追加写入；`ERROR_WHEN_DATA_EXISTS`
表示集合已有数据时直接报错。 |
+| common-options | - | 否 | - | 通用 Sink
插件参数，详见 [Sink Common Options](../common-options/sink-common-options.md)。 |
### 提示
-> 1. MongoDB Sink
的连接器级数据刷新逻辑由以下三个参数共同控制：`buffer-flush.max-rows`、`buffer-flush.interval` 和
`checkpoint.interval`。<br/>
- > 任一条件满足时，都会触发数据刷写。<br/>
-> 2. 兼容历史参数 `upsert-key`。若已设置 `upsert-key`，请勿同时设置 `primary-key`。
+> 1. MongoDB Sink 的连接器级数据刷新由三个参数共同控制：`buffer-flush.max-rows`、
+> `buffer-flush.interval` 和 `checkpoint.interval`。任一条件触发都会立刻刷写。
+> 2. 兼容历史参数 `upsert-key` 作为 `primary-key` 的回退名。若已设置 `upsert-key`，请勿同时
+> 设置 `primary-key`。
+> 3. `transaction` 选项与下文 Zeta 定时刷新互斥，请二选一。
### Zeta 定时刷新
-该引擎级能力仅由 Zeta 支持，Spark 和 Flink 不会注入 `FlushSignal`。在 Zeta 中，可以在 `env` 块配置
`sink.flush.interval`，使未达到 `buffer-flush.max-rows` 的待处理 bulk 请求也能定时写出。与
`buffer-flush.interval` 不同，引擎定时器不需要新记录到达才触发检查。
+该引擎级能力仅由 Zeta 支持，Spark 和 Flink 不会注入 `FlushSignal` 记录。在 Zeta 中可以在
+`env` 块配置 `sink.flush.interval`，使未达到 `buffer-flush.max-rows` 的待处理 bulk 请求也能
+定时刷写出去。和 `buffer-flush.interval` 不同，引擎定时器不依赖新记录到达即可触发检查。
-定时刷新仅在 `transaction = false` 时启用。MongoDB 事务模式通过 checkpoint
提交，因此会禁用定时刷新以保持事务边界。初始定时刷新实现提供至少一次语义，不提供基于 2PC 的精确一次语义。启用 upsert
并使用确定性主键可使重试具备幂等性。
+定时刷新仅在 `transaction = false` 时启用。MongoDB 事务模式通过 checkpoint 提交，因此会禁用
+定时刷新以保持事务边界。初始定时刷新实现提供至少一次语义，不提供基于 2PC 的精确一次语义。
+启用 upsert 并使用确定性主键可使重试具备幂等性。
```hocon
env {
@@ -111,29 +130,28 @@ sink {
## 如何创建 MongoDB 数据同步任务
-下面示例展示了一个将随机生成的数据写入 MongoDB 的数据同步任务：
+下面示例展示了一个将随机生成的数据写入 MongoDB 的任务：
-```bash
-# 设置作业的基本配置
+```hocon
env {
parallelism = 1
job.mode = "BATCH"
- checkpoint.interval = 1000
+ checkpoint.interval = 1000
}
source {
FakeSource {
- row.num = 2
- bigint.min = 0
- bigint.max = 10000000
- split.num = 1
- split.read-interval = 300
- schema {
- fields {
- c_bigint = bigint
- }
+ row.num = 2
+ bigint.min = 0
+ bigint.max = 10000000
+ split.num = 1
+ split.read-interval = 300
+ schema {
+ fields {
+ c_bigint = bigint
}
}
+ }
}
sink {
@@ -147,7 +165,7 @@ sink {
### 多表写入
-当上游记录带有表元数据时，`database` 和 `collection` 可以使用占位符。常用占位符包括
+当上游记录携带表元数据时，`database` 和 `collection` 可以使用占位符。常用占位符包括
`${database_name}`、`${schema_name}` 和 `${table_name}`。
```hocon
@@ -190,7 +208,7 @@ source {
sink {
MongoDB {
- uri = "mongodb://user:[email protected]:27017/test_db?retryWrites=true"
+ uri = "mongodb://127.0.0.1:27017/test_db?retryWrites=true"
database = "test_db"
collection = "${database_name}_${schema_name}_${table_name}_check"
}
@@ -204,31 +222,31 @@ sink {
无认证的单节点连接：
```bash
-mongodb://127.0.0.0:27017/mydb
+mongodb://127.0.0.1:27017/mydb
```
副本集连接：
```bash
-mongodb://127.0.0.0:27017/mydb?replicaSet=xxx
+mongodb://127.0.0.1:27017/mydb?replicaSet=xxx
```
带认证的副本集连接：
```bash
-mongodb://admin:[email protected]:27017/mydb?replicaSet=xxx&authSource=admin
+mongodb://admin:[email protected]:27017/mydb?replicaSet=xxx&authSource=admin
```
多节点副本集连接：
```bash
-mongodb://127.0.0.1:27017,127.0.0.2:27017,127.0.0.3:27017/mydb?replicaSet=xxx
+mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb?replicaSet=xxx
```
-分片集群连接：
+分片集群连接（通过一个 `mongos` 路由）：
```bash
-mongodb://127.0.0.0:27017/mydb
+mongodb://mongos1.example.com:27017,mongos2.example.com:27017,mongos3.example.com:27017/mydb
```
多个 mongos 节点连接：
@@ -237,11 +255,11 @@ mongodb://127.0.0.0:27017/mydb
mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb
```
-注意：URI 中的用户名与密码在拼接前必须进行 URL 编码。
+> 注意：URI 中的用户名与密码在拼接前必须进行 URL 编码。
### Buffer Flush 示例
-```bash
+```hocon
sink {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
@@ -256,32 +274,32 @@ sink {
### 为什么不推荐频繁使用事务？
虽然 MongoDB 自 4.2 版本起已完全支持多文档事务，但这并不意味着所有场景都应使用。
-事务意味着加锁、节点协调、额外开销和性能损耗。
-设计系统时应遵循的原则是：**能不用事务就不要用事务**。
-合理的系统设计可以在大多数情况下避免对事务的依赖。
+事务意味着加锁、节点协调、额外往返和性能损耗。设计系统时应遵循的原则是：**能不用事务就
+不要用事务**。合理的系统设计可以在大多数情况下避免对事务的依赖。
### 幂等写入（Idempotent Writes）
通过定义明确的主键并启用 `upsert` 模式，可以实现精准一次写入（exactly-once）语义。
-当配置中定义了 `primary-key` 且启用了 `upsert-enable`，MongoDB Sink 将使用 Upsert 语义而非普通
INSERT 语句。
-SeaTunnel 会将定义的主键作为 MongoDB 的复合主键，在 Upsert 模式下进行写入，以确保幂等性。
+当配置中定义了 `primary-key` 且启用了 `upsert-enable`，MongoDB Sink 将使用 Upsert 语义而非
+普通 INSERT 语句。SeaTunnel 会将定义的主键作为 MongoDB 的复合主键，在 Upsert 模式下写入，
+以确保幂等性。
-若作业在运行过程中失败，SeaTunnel 会从上一个成功的 checkpoint 恢复并重新处理数据，这可能导致重复数据。
-强烈建议启用 Upsert 模式，以避免主键冲突或重复插入。
+若作业在运行过程中失败，SeaTunnel 会从上一个成功的 checkpoint 恢复并重新处理数据，这可能
+导致重复数据。强烈建议启用 Upsert 模式，以避免主键冲突或重复插入。
-```bash
+```hocon
sink {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
database = "test_db"
collection = "users"
upsert-enable = true
- primary-key = ["name","status"]
+ primary-key = ["name", "status"]
}
}
```
## 更新日志
-<ChangeLog />
+<ChangeLog />
\ No newline at end of file
diff --git a/docs/zh/connectors/source/MongoDB.md
b/docs/zh/connectors/source/MongoDB.md
index 490dd694a6..2f81096a2c 100644
--- a/docs/zh/connectors/source/MongoDB.md
+++ b/docs/zh/connectors/source/MongoDB.md
@@ -13,48 +13,58 @@ import ChangeLog from '../changelog/connector-mongodb.md';
## 关键特性
- [x] [批](../../introduction/concepts/connector-v2-features.md)
-- [ ] [流](../../introduction/concepts/connector-v2-features.md)
+- [x] [流](../../introduction/concepts/connector-v2-features.md)
- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
- [x] [列投影](../../introduction/concepts/connector-v2-features.md)
- [x] [并行性](../../introduction/concepts/connector-v2-features.md)
-- [x] [支持用户自定义split](../../introduction/concepts/connector-v2-features.md)
+- [x] [支持用户自定义 split](../../introduction/concepts/connector-v2-features.md)
## 描述
-MongoDB连接器提供了从MongoDB读取数据和向MongoDB写入数据的能力。
-本文档描述了如何设置MongoDB连接器以对MongoDB运行数据读取。
+MongoDB 源连接器从 MongoDB 集合中读取文档，并把每个 BSON 文档转换为 SeaTunnel 行记录。
+它同时支持**批**和**流**两种作业模式，并通过 `partition.split-key` 把集合按取值范围切分为多个
+split，实现并行读取。
+
+在不扫描整张集合的前提下，可以缩小读取范围并控制返回字段：
+
+- 使用 `match.query` 过滤符合条件的文档。
+- 使用 `match.projection` 控制结果中返回的字段。
+- 使用 `flat.sync-string` 把整篇文档作为一条 JSON `STRING` 列读入，跳过固定 schema 的定义。
+
+在流模式下，连接器读取已分配的 split，并通过 checkpoint 跟踪读取位置；任务重启后会从上次提交
+的游标继续读取。
## 支持的数据源信息
-为了使用Mongodb连接器，需要以下依赖关系。
-它们可以通过install-plugin.sh或Maven中央存储库下载。
+要使用 MongoDB 连接器，需要以下依赖。可以通过 `install-plugin.sh` 或 Maven 中央仓库下载。
| 数据源 | 支持的版本 | 依赖
|
-|------------|--------------------|---------------------------------------------------------------------------------------|
-| MongoDB | universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-mongodb)
|
+|--------|------------|-----------------------------------------------------------------------------------------|
+| MongoDB | 通用版本 |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-mongodb)
|
## 数据类型映射
-下表列出了从MongoDB BSON类型到SeaTunnel数据类型的字段数据类型映射。
-
-| MongoDB BSON type | SeaTunnel 数据类型 |
-|-------------------|----------------|
-| ObjectId | STRING |
-| String | STRING |
-| Boolean | BOOLEAN |
-| Binary | BINARY |
-| Int32 | INTEGER |
-| Int64 | BIGINT |
-| Double | DOUBLE |
-| Decimal128 | DECIMAL |
-| Date | Date |
-| Timestamp | Timestamp |
-| Object | ROW |
-| Array | ARRAY |
-
-对于MongoDB中的特定类型，我们使用扩展JSON格式将其映射到SeaTunnel STRING类型。
-
-| MongoDB BSON type | SeaTunnel STRING
|
+下表列出了从 MongoDB BSON 类型到 SeaTunnel 数据类型的字段映射。
+
+| MongoDB BSON 类型 | SeaTunnel 数据类型 |
+|-------------------|-------------------|
+| ObjectId | STRING |
+| String | STRING |
+| Boolean | BOOLEAN |
+| Binary | BINARY |
+| Int32 | INTEGER |
+| Int64 | BIGINT |
+| Double | DOUBLE |
+| Decimal128 | DECIMAL |
+| Date | Date |
+| Timestamp | Timestamp |
+| Object | ROW |
+| Array | ARRAY |
+
+针对 MongoDB 中的特殊类型，连接器使用扩展 JSON（Extended JSON）格式映射到 SeaTunnel 的
+`STRING` 类型。
+
+| MongoDB BSON 类型 | SeaTunnel STRING
|
|-------------------|----------------------------------------------------------------------------------------------|
| Symbol | {"_value": {"$symbol": "12"}}
|
| RegularExpression | {"_value": {"$regularExpression": {"pattern": "^9$",
"options": "i"}}} |
@@ -63,42 +73,43 @@ MongoDB连接器提供了从MongoDB读取数据和向MongoDB写入数据的能
**提示**
-> 1.在SeaTunnel中使用DECIMAL类型时，请注意最大范围不能超过34位数字，这意味着您应该使用DECIMAL(34,18)。<br/>
+> 1. 在 SeaTunnel 中使用 `DECIMAL` 类型时，最大精度不能超过 34 位。建议使用
+> `decimal(34, 18)` 以满足支持的精度与标度。
## 源配置项
-| 参数名 | 类型 | 必须 | 默认值 | 描述
|
-|----------------------|---------|----|------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| uri | String | 是 | - | MongoDB标准连接uri。例如
mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true.
|
-| database | String | 是 | - |
要读取或写入的MongoDB数据库的名称。
|
-| collection | String | 是 | - |
要读取或写入的MongoDB集合的名称。
|
-| schema | Config | 是 | - | MongoDB 的 BSON 与
SeaTunnel 数据结构映射。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
|
-| match.query | String | 否 | - |
在MongoDB中，过滤器用于过滤查询操作的文档。
|
-| match.projection | String | 否 | - |
在MongoDB中，投影用于控制查询结果中包含的字段。
|
-| partition.split-key | String | 否 | _id | 分片字段。
|
-| partition.split-size | Long | 否 | 64 * 1024 * 1024 | 分片大小。
|
-| cursor.no-timeout | Boolean | 否 | true |
MongoDB服务器通常在非活动期（10分钟）后超时空闲游标，以防止过度使用内存。将此选项设置为true以防止这种情况发生。但是，如果应用程序处理当前一批文档的时间超过30分钟，则会话将标记为已过期并关闭。
|
-| fetch.size | Int | 否 | 2048 |
设置每批从服务器获取的文档数量。设置适当的批大小可以提高查询性能，避免一次获取大量数据造成的内存压力。
|
-| max.time-min | Long | 否 | 10 |
此参数是一个MongoDB查询选项，用于限制查询操作的最大执行时间。maxTimeMin的值以分钟为单位。如果查询的执行时间超过指定的时间限制，MongoDB将终止操作并返回错误。
|
-| flat.sync-string | Boolean | 否 | false | 开启后，会把整个 MongoDB
文档映射到一个 SeaTunnel `STRING` 字段。此时 schema 只能配置一个字段，并且该字段类型必须是 `STRING`。
|
-| common-options | | 否 | - | 源插件常用参数，请参考
[源通用选项](../common-options/source-common-options.md)
|
+| 参数名称 | 类型 | 是否必填 | 默认值 | 描述
|
+|-----------------------|---------|----------|--------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| uri | String | 是 | - | MongoDB
标准连接 URI，例如
`mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true`。更多示例请参考
[参数说明](#参数说明)。
|
+| database | String | 是 | - | 要读取的
MongoDB 数据库名称。
|
+| collection | String | 是 | - | 要读取的
MongoDB 集合名称。
|
+| schema | Config | 是 | - | MongoDB 的
BSON 与 SeaTunnel 数据结构的映射。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
|
+| match.query | String | 否 | - | 用于过滤读取文档的
MongoDB 查询表达式。兼容旧版参数名 `matchQuery`。
|
+| match.projection | String | 否 | - |
用于控制查询结果中包含字段的 MongoDB 投影表达式。
|
+| partition.split-key | String | 否 | _id | 用作 MongoDB
分片字段的列名，连接器会按该字段的取值范围切分集合。
|
+| partition.split-size | Long | 否 | 64 * 1024 * 1024 | 每个 MongoDB
split 的大小。split 越小并行度越高，split 越大并行度越低。
|
+| cursor.no-timeout | Boolean | 否 | true | MongoDB
服务端默认会在游标空闲 10 分钟后关闭游标以回收内存。将此选项设置为 `true` 可让游标在长时间运行的批次中保持打开。如果应用持有批次超过 30
分钟，MongoDB 会将当前会话标记为过期并关闭。
|
+| fetch.size | Int | 否 | 2048 |
每批从服务器获取的文档数。合理设置可以提升查询性能并降低一次性获取大量数据带来的内存压力。
|
+| max.time-min | Long | 否 | 10 | 每次 MongoDB
查询的最大执行时间（分钟）。超过该限制 MongoDB 将终止操作并返回错误。
|
+| flat.sync-string | Boolean | 否 | false | 开启后，连接器会把整篇
MongoDB 文档映射到一个 SeaTunnel `STRING` 字段。此时 schema 只能声明一个字段，且该字段必须是 `STRING` 类型。
|
+| common-options | | 否 | - | 源插件通用参数，详见
[源通用选项](../common-options/source-common-options.md)。
|
### 提示
-> 1.参数`match.query `与历史旧版本参数`matchQuery `兼容，它们是等效的替换。<br/>
+> 1. `match.query` 与旧版参数名 `matchQuery` 等价，二者不能同时设置。
+> 2. 使用 `partition.split-key` 时建议选择有索引的字段，能显著加快 split 边界扫描。
+> 3. 当 `flat.sync-string = true` 时，schema 仅用于声明单个接收文档的 `STRING` 字段。
-## 如何创建MongoDB数据同步作业
+## 如何创建 MongoDB 数据同步作业
-以下示例演示了如何创建数据同步作业，该作业从MongoDB读取数据并将其打印到本地客户端：
+下面的示例从 MongoDB 读取数据并打印到本地客户端：
-```bash
-# 设置要执行的任务的基本配置
+```hocon
env {
parallelism = 1
job.mode = "BATCH"
}
-# 创建MongoDB源
source {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
@@ -135,7 +146,6 @@ source {
}
}
-# 控制台打印读取的Mongodb数据
sink {
Console {
parallelism = 1
@@ -145,9 +155,9 @@ sink {
## 参数说明
-### MongoDB数据库连接URI示例
+### MongoDB 数据库连接 URI 示例
-未经身份验证的单节点连接：
+无认证的单节点连接：
```bash
mongodb://192.168.0.100:27017/mydb
@@ -159,7 +169,7 @@ mongodb://192.168.0.100:27017/mydb
mongodb://192.168.0.100:27017/mydb?replicaSet=xxx
```
-经过身份验证的副本集连接：
+带认证的副本集连接：
```bash
mongodb://admin:[email protected]:27017/mydb?replicaSet=xxx&authSource=admin
@@ -171,26 +181,26 @@
mongodb://admin:[email protected]:27017/mydb?replicaSet=xxx&authSource=admi
mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb?replicaSet=xxx
```
-分片集群连接：
+分片集群连接（通过一个 `mongos` 路由）：
```bash
-mongodb://192.168.0.100:27017/mydb
+mongodb://mongos1.example.com:27017,mongos2.example.com:27017,mongos3.example.com:27017/mydb
```
-多个mongos连接：
+多个 mongos 节点连接：
```bash
mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb
```
-注意：URI中的用户名和密码在连接到连接字符串之前必须进行URL编码。
+> 注意：URI 中的用户名与密码在拼接到连接字符串前必须进行 URL 编码。
### 匹配查询扫描
-在数据同步场景中，需要尽早使用matchQuery方法来减少后续操作员需要处理的文档数量，从而提高性能。
-下面是一个使用`match.query的seatunnel的简单示例`
+在数据同步场景中，建议尽早使用 `match.query` 减少下游算子需要处理的文档数量，从而提升整体
+性能。下面是一个简单的示例：
-```bash
+```hocon
source {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
@@ -207,38 +217,38 @@ source {
}
```
-以下是各种数据类型的MatchQuery查询语句的示例：
+下面是常见数据类型对应的 `match.query` 表达式：
```bash
-# Query Boolean type
-"{c_boolean:true}"
-# Query string type
-"{c_string:\"OCzCj\"}"
-# Query the integer
-"{c_int:2}"
-# Type of query time
-"{c_date:ISODate(\"2023-06-26T16:00:00.000Z\")}"
-# Query floating point type
-{c_double:{$gte:1.71763202185342e+308}}
+# 布尔类型
+"{c_boolean: true}"
+# 字符串类型
+"{c_string: \"OCzCj\"}"
+# 整数类型
+"{c_int: 2}"
+# 日期类型
+"{c_date: {\$date: \"2023-06-26T16:00:00.000Z\"}}"
+# 浮点类型
+"{c_double: {\$gte: 1.71763202185342e+308}}"
```
-请参阅如何编写`match.query的语法`：https://www.mongodb.com/docs/manual/tutorial/query-documents
+完整查询语法请参考 MongoDB 官方文档：
+<https://www.mongodb.com/docs/manual/tutorial/query-documents>
### 投影扫描
-在MongoDB中，Projection用于控制查询结果中包含哪些字段。这可以通过指定哪些字段需要返回，哪些字段不需要返回来实现。
-在find（）方法中，投影对象可以作为第二个参数传递。投影对象的键表示要包含或排除的字段，值1表示包含，0表示排除。
-这里有一个简单的例子，假设我们有一个名为users的集合：
+MongoDB 中的 Projection 用来控制查询结果中返回哪些字段：在 `find()` 方法中通过第二个参数
+传入一个投影对象，键表示字段，值 `1` 表示包含，`0` 表示排除。例如对于 `users` 集合：
-```bash
-# Returns only the name and email fields
+```javascript
+// 仅返回 `name` 字段，过滤掉 `email` 字段
db.users.find({}, { name: 1, email: 0 });
```
-在数据同步场景中，需要尽早使用投影来减少后续操作员需要处理的文档数量，从而提高性能。
-以下是一个使用投影的seatunnel的简单示例：
+在数据同步场景中，尽早使用 Projection 可以减少下游算子需要处理的字段数量。下面是 SeaTunnel
+中使用投影的简单示例：
-```bash
+```hocon
source {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
@@ -252,15 +262,15 @@ source {
}
}
}
-
```
### 分区扫描
-为了加快并行源任务实例中的数据读取速度，seatunnel为MongoDB集合提供了分区扫描功能。提供了以下分区策略。
-用户可以通过设置用于分片字段的partition.split-key和用于分片大小的partition.split-size来控制数据分片。
+为了加速并行源任务中的数据读取，SeaTunnel 为 MongoDB 集合提供了分区扫描能力。通过
+`partition.split-key` 指定分片字段、`partition.split-size` 指定每个 split 的大小，可以控制
+数据分片方式：
-```bash
+```hocon
source {
MongoDB {
uri = "mongodb://user:[email protected]:27017"
@@ -276,17 +286,18 @@ source {
}
}
}
-
```
+> 建议选择有索引的字段作为 split key，能显著加快 split 边界扫描。
+
### Flat Sync String
-通过使用“flat.sync string”，只能设置一个字段属性值，并且字段类型必须是string。
-此操作将对单个MongoDB数据条目执行字符串映射。
+启用 `flat.sync-string` 后，只需声明一个 `STRING` 类型字段，连接器会把每条 MongoDB 文档序列化
+为扩展 JSON 字符串写入该字段。
-```bash
+```hocon
env {
- parallelism = 10
+ parallelism = 1
job.mode = "BATCH"
}
source {
@@ -307,147 +318,41 @@ sink {
}
```
-使用与修改后的参数同步的数据样本，例如：
+通过该配置写入的样本数据示例：
```json
{
- "_id":{
- "$oid":"643d41f5fdc6a52e90e59cbf"
+ "_id": {
+ "$oid": "643d41f5fdc6a52e90e59cbf"
},
- "c_map":{
- "OQBqH":"jllt",
- "rkvlO":"pbfdf",
- "pCMEX":"hczrdtve",
- "DAgdj":"t",
- "dsJag":"voo"
+ "c_map": {
+ "OQBqH": "jllt",
+ "rkvlO": "pbfdf",
+ "pCMEX": "hczrdtve",
+ "DAgdj": "t",
+ "dsJag": "voo"
},
- "c_array":[
- {
- "$numberInt":"-865590937"
- },
- {
- "$numberInt":"833905600"
- },
- {
- "$numberInt":"-1104586446"
- },
- {
- "$numberInt":"2076336780"
- },
- {
- "$numberInt":"-1028688944"
- }
+ "c_array": [
+ { "$numberInt": "-865590937" },
+ { "$numberInt": "833905600" },
+ { "$numberInt": "-1104586446" },
+ { "$numberInt": "2076336780" },
+ { "$numberInt": "-1028686444" }
],
- "c_string":"bddkzxr",
- "c_boolean":false,
- "c_tinyint":{
- "$numberInt":"39"
- },
- "c_smallint":{
- "$numberInt":"23672"
- },
- "c_int":{
- "$numberInt":"-495763561"
- },
- "c_bigint":{
- "$numberLong":"3768307617923954543"
- },
- "c_float":{
- "$numberDouble":"5.284220288280258E37"
- },
- "c_double":{
- "$numberDouble":"1.1706091642478246E308"
- },
- "c_bytes":{
- "$binary":{
- "base64":"ZWJ4",
- "subType":"00"
- }
- },
- "c_date":{
- "$date":{
- "$numberLong":"1686614400000"
- }
- },
- "c_decimal":{
- "$numberDecimal":"683265300"
- },
- "c_timestamp":{
- "$date":{
- "$numberLong":"1684283772000"
- }
- },
- "c_row":{
- "c_map":{
- "OQBqH":"cbrzhsktmm",
- "rkvlO":"qtaov",
- "pCMEX":"tuq",
- "DAgdj":"jzop",
- "dsJag":"vwqyxtt"
- },
- "c_array":[
- {
- "$numberInt":"1733526799"
- },
- {
- "$numberInt":"-971483501"
- },
- {
- "$numberInt":"-1716160960"
- },
- {
- "$numberInt":"-919976360"
- },
- {
- "$numberInt":"727499700"
- }
- ],
- "c_string":"oboislr",
- "c_boolean":true,
- "c_tinyint":{
- "$numberInt":"-66"
- },
- "c_smallint":{
- "$numberInt":"1308"
- },
- "c_int":{
- "$numberInt":"-1573886733"
- },
- "c_bigint":{
- "$numberLong":"4877994302999518682"
- },
- "c_float":{
- "$numberDouble":"1.5353209063652051E38"
- },
- "c_double":{
- "$numberDouble":"1.1952441956458565E308"
- },
- "c_bytes":{
- "$binary":{
- "base64":"cWx5Ymp0Yw==",
- "subType":"00"
- }
- },
- "c_date":{
- "$date":{
- "$numberLong":"1686614400000"
- }
- },
- "c_decimal":{
- "$numberDecimal":"656406177"
- },
- "c_timestamp":{
- "$date":{
- "$numberLong":"1684283772000"
- }
- }
- },
- "id":{
- "$numberInt":"2"
- }
+ "c_string": "bddkzxr",
+ "c_boolean": false,
+ "c_tinyint": { "$numberInt": "39" },
+ "c_smallint": { "$numberInt": "23672" },
+ "c_int": { "$numberInt": "-495763561" },
+ "c_bigint": { "$numberLong": "3768307617923954543" },
+ "c_double": { "$numberDouble": "1.1706091642478246E308" },
+ "c_bytes": { "$binary": { "base64": "ZWJ4", "subType": "00" } },
+ "c_date": { "$date": { "$numberLong": "1686614400000" } },
+ "c_decimal": { "$numberDecimal": "683265300" },
+ "c_timestamp": { "$date": { "$numberLong": "1684283772000" } }
}
```
## 修改日志
-<ChangeLog />
+<ChangeLog />
\ No newline at end of file
