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 8783b50858 [Docs][Connector-V2] Improve InfluxDB Jdbc and MongoDB docs
(#11308)
8783b50858 is described below
commit 8783b50858a519ee170ec64b8c3396aa48d830c1
Author: Daniel Carter <[email protected]>
AuthorDate: Sat Jul 4 09:50:56 2026 +0800
[Docs][Connector-V2] Improve InfluxDB Jdbc and MongoDB docs (#11308)
Co-authored-by: DanielCarter-stack
<[email protected]>
---
docs/en/connectors/sink/InfluxDB.md | 4 +++
docs/en/connectors/sink/Jdbc.md | 28 +++++++++++++++
docs/en/connectors/sink/MongoDB.md | 66 +++++++++++++++++++++++++++++++----
docs/en/connectors/source/InfluxDB.md | 5 +--
docs/en/connectors/source/Jdbc.md | 6 ++++
docs/en/connectors/source/MongoDB.md | 9 +++--
docs/zh/connectors/sink/InfluxDB.md | 3 ++
docs/zh/connectors/sink/Jdbc.md | 33 ++++++++++++++++++
docs/zh/connectors/sink/MongoDB.md | 66 +++++++++++++++++++++++++++++++----
docs/zh/connectors/source/InfluxDB.md | 5 +--
docs/zh/connectors/source/Jdbc.md | 6 ++++
docs/zh/connectors/source/MongoDB.md | 9 +++--
12 files changed, 212 insertions(+), 28 deletions(-)
diff --git a/docs/en/connectors/sink/InfluxDB.md
b/docs/en/connectors/sink/InfluxDB.md
index 98a5b58a6c..dbe322165c 100644
--- a/docs/en/connectors/sink/InfluxDB.md
+++ b/docs/en/connectors/sink/InfluxDB.md
@@ -54,6 +54,7 @@ The name of `influxDB` database
The name of `influxDB` measurement. This option is optional. If it is omitted,
the sink uses the
input table full name as the measurement name, which is useful for multi-table
writes.
+For multi-table input, make sure the generated table names are valid InfluxDB
measurement names.
### username [string]
@@ -180,6 +181,9 @@ sink {
### Multiple Table
+When `measurement` is omitted, each upstream table is written to a measurement
named after that
+table. This is the usual setting for multi-table input.
+
```hocon
env {
parallelism = 1
diff --git a/docs/en/connectors/sink/Jdbc.md b/docs/en/connectors/sink/Jdbc.md
index 4e0d9e6503..dccb289488 100644
--- a/docs/en/connectors/sink/Jdbc.md
+++ b/docs/en/connectors/sink/Jdbc.md
@@ -46,6 +46,8 @@ support `Xa transactions`. You can set `is_exactly_once=true`
to enable it.
| tableSuffix | String | No | -
|
| primary_keys | Array | No | -
|
| connection_check_timeout_sec | Int | No | 30
|
+| connect_timeout_ms | Int | No | 86400000
|
+| socket_timeout_ms | Int | No | 86400000
|
| max_retries | Int | No | 0
|
| batch_size | Int | No | 1000
|
| batch_interval_ms | Long | No | 0
|
@@ -62,10 +64,16 @@ support `Xa transactions`. You can set
`is_exactly_once=true` to enable it.
| data_save_mode | Enum | No | APPEND_DATA
|
| custom_sql | String | No | -
|
| enable_upsert | Boolean | No | true
|
+| is_primary_key_updated | Boolean | No | true
|
+| support_upsert_by_insert_only | Boolean | No | false
|
| table_options | Map | No | -
|
| use_copy_statement | Boolean | No | false
|
| oracle_insert_mode | Enum | No |
CONVENTIONAL |
| create_index | Boolean | No | true
|
+| use_kerberos | Boolean | No | false
|
+| kerberos_principal | String | No | -
|
+| kerberos_keytab_path | String | No | -
|
+| krb5_path | String | No |
/etc/krb5.conf |
| access_key_id | String | No |
|
| secret_access_key | String | No |
|
| region | String | No |
|
@@ -163,6 +171,14 @@ This option is used to support operations such as
`insert`, `delete`, and `updat
The time in seconds to wait for the database operation used to validate the
connection to complete.
+### connect_timeout_ms [int]
+
+Connection timeout in milliseconds when establishing the JDBC connection. The
default is 24 hours. Set it to `0` to disable the timeout.
+
+### socket_timeout_ms [int]
+
+Socket read timeout in milliseconds after the JDBC connection is established.
The default is 24 hours. Set it to `0` to disable the timeout.
+
### max_retries [int]
The number of retries to submit failed (executeBatch)
@@ -283,6 +299,14 @@ The generated `CREATE TABLE` statement appends `ENGINE`,
`DEFAULT CHARSET`, and
Enable upsert by primary_keys exist, If the task has no key duplicate data,
setting this parameter to `false` can speed up data import
+### is_primary_key_updated [boolean]
+
+Whether primary key fields are included when generating update statements.
Keep the default unless your target database requires primary key columns to be
skipped during updates.
+
+### support_upsert_by_insert_only [boolean]
+
+Whether to support upsert behavior through insert-only statements for
compatible dialects. This is an advanced compatibility option and is disabled
by default.
+
### use_copy_statement [boolean]
Use `COPY ${table} FROM STDIN` statement to import data. Only drivers with
`getCopyAPI()` method connections are supported. e.g.: Postgresql driver
`org.postgresql.Driver`.
@@ -307,6 +331,10 @@ Create the index(contains primary key and any other
indexes) or not when auto-cr
Notice: Note that this will sacrifice read performance, so you'll need to
manually create indexes after the table migration to improve read performance
+### use_kerberos [boolean]
+
+Whether to enable Kerberos authentication for JDBC connections. When enabled,
also configure `kerberos_principal`, `kerberos_keytab_path`, and `krb5_path` as
required by your environment.
+
### access_key_id [String]
The access_key_id in AWS authentication. Only valid for dialect="dsql"
diff --git a/docs/en/connectors/sink/MongoDB.md
b/docs/en/connectors/sink/MongoDB.md
index c03d59a8c9..dc603292f5 100644
--- a/docs/en/connectors/sink/MongoDB.md
+++ b/docs/en/connectors/sink/MongoDB.md
@@ -67,16 +67,16 @@ The following table lists the field data type mapping from
MongoDB BSON type to
|-----------------------|----------|----------|--------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| 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 `${table_name}`,`${schema_name}`,`${table_name}` as
placeholders, for example: `collection =
"${database_name}_${schema_name}_${table_name}_check"` |
-| buffer-flush.max-rows | String | No | 1000 | Specifies the maximum
number of buffered rows per batch request.
|
-| buffer-flush.interval | String | No | 30000 | Specifies the maximum
interval of buffered rows per batch request, the unit is millisecond.
|
-| retry.max | String | No | 3 | Specifies the max
number of retry if writing records to database failed.
|
-| retry.interval | Duration | No | 1000 | Specifies the retry
time interval if writing records to database failed, the unit is millisecond.
|
+| 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 | String | No | APPEND_DATA | The data
saving mode of mongodb，Option introduction,`DROP_DATA`:The collection will be
cleared before inserting data;`APPEND_DATA`:Append data
;`ERROR_WHEN_DATA_EXISTS`:An error will be reported if there is data in the
collection. |
+| data_save_mode | Enum | No | APPEND_DATA | The data
saving mode of mongodb. Supported values: `DROP_DATA`, `APPEND_DATA`, and
`ERROR_WHEN_DATA_EXISTS`. |
### Tips
@@ -114,13 +114,65 @@ source {
sink {
MongoDB{
- uri = mongodb://user:[email protected]:27017
+ uri = "mongodb://user:[email protected]:27017"
database = "test"
collection = "test"
}
}
```
+### 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}`.
+
+```hocon
+source {
+ FakeSource {
+ tables_configs = [
+ {
+ schema = {
+ table = "testDatabase1.testSchema1.testTable1"
+ fields {
+ id = int
+ value = string
+ }
+ }
+ rows = [
+ {
+ kind = INSERT
+ fields = [1, "NEW"]
+ }
+ ]
+ },
+ {
+ schema = {
+ table = "testDatabase2.testSchema2.testTable2"
+ fields {
+ id = int
+ amount = "decimal(16, 1)"
+ }
+ }
+ rows = [
+ {
+ kind = INSERT
+ fields = [1, 6.3]
+ }
+ ]
+ }
+ ]
+ }
+}
+
+sink {
+ MongoDB {
+ uri = "mongodb://user:[email protected]:27017/test_db?retryWrites=true"
+ database = "test_db"
+ collection = "${database_name}_${schema_name}_${table_name}_check"
+ }
+}
+```
+
## Parameter Interpretation
### MongoDB Database Connection URI Examples
diff --git a/docs/en/connectors/source/InfluxDB.md
b/docs/en/connectors/source/InfluxDB.md
index 0273f8bf71..0bab3502d5 100644
--- a/docs/en/connectors/source/InfluxDB.md
+++ b/docs/en/connectors/source/InfluxDB.md
@@ -19,7 +19,7 @@ query and an optional parallel scan mode that splits one
query by an integer col
supports query SQL and can achieve projection effect.
- [x] [parallelism](../../introduction/concepts/connector-v2-features.md)
-- [ ] [support user-defined
split](../../introduction/concepts/connector-v2-features.md)
+- [x] [support user-defined
split](../../introduction/concepts/connector-v2-features.md)
## Options
@@ -35,7 +35,7 @@ supports query SQL and can achieve projection effect.
| upper_bound | int | no | - | Upper bound of
`split_column` when parallel scan is enabled.
|
| partition_num | int | no | 0 | Number of query
splits. `0` means the source runs the original `sql` as one split.
|
| split_column | string | no | - | Integer column used
to split the query when parallel scan is enabled. |
-| where | string | no | - | Reserved source
option. The current split logic reads the `where` keyword from `sql` directly.
|
+| where | string | no | - | Reserved source
option. The current split logic reads the lowercase `where` keyword from `sql`
directly. |
| epoch | string | no | n | Time precision
returned by InfluxDB. For example: `H`, `m`, `s`, `MS`, `u`, `n`.
|
| connect_timeout_ms | long | no | 15000 | Timeout for
connecting to InfluxDB, in milliseconds.
|
| query_timeout_sec | int | no | 3 | Timeout for
querying InfluxDB, in seconds.
|
@@ -94,6 +94,7 @@ The column used to split one query into multiple range
queries.
> - influxDB time is not supported as a segmented primary key because the time
> field cannot participate in mathematical calculation
> - Currently, `split_column` only supports integer data segmentation, and
> does not support `float`, `string`, `date` and other types.
> - `split_column`, `lower_bound`, `upper_bound`, and `partition_num` must be
> configured together.
+> - If the split query contains a filter, use lowercase `where` in `sql`, for
example `select * from test where age > 0`. The current split parser is
case-sensitive.
### upper_bound [int]
diff --git a/docs/en/connectors/source/Jdbc.md
b/docs/en/connectors/source/Jdbc.md
index 88284fc874..e08e25495e 100644
--- a/docs/en/connectors/source/Jdbc.md
+++ b/docs/en/connectors/source/Jdbc.md
@@ -51,6 +51,8 @@ supports query SQL and can achieve projection effect.
| compatible_mode | String | No | -
| The compatible mode of database, required when the database supports
multiple compatible modes.<br/> For example, when using OceanBase database, you
need to set it to 'mysql' or 'oracle'. <br/> when using starrocks, you need set
it to `starrocks`
[...]
| dialect | String | No | -
| The appointed dialect, if it does not exist, is still obtained
according to the url, and the priority is higher than the url. <br/> For
example,when using starrocks, you need set it to `starrocks`
[...]
| connection_check_timeout_sec | Int | No | 30
| The time in seconds to wait for the database operation used to validate
the connection to complete.
[...]
+| connect_timeout_ms | Int | No | 86400000
| Connection timeout in milliseconds when establishing the JDBC
connection. `0` means no timeout.
[...]
+| socket_timeout_ms | Int | No | 86400000
| Socket read timeout in milliseconds after the JDBC connection is
established. `0` means no timeout.
[...]
| partition_column | String | No | -
| The column name for split data.
[...]
| partition_upper_bound | Long | No | -
| The partition_column max value for scan, if not set SeaTunnel will
query database get max value.
[...]
| partition_lower_bound | Long | No | -
| The partition_column min value for scan, if not set SeaTunnel will
query database get min value.
[...]
@@ -58,6 +60,10 @@ supports query SQL and can achieve projection effect.
| decimal_type_narrowing | Boolean | No | true
| Decimal type narrowing, if true, the decimal type will be narrowed to
the int or long type if without loss of precision. Only support for Oracle at
now. Please refer to `decimal_type_narrowing` below
[...]
| int_type_narrowing | Boolean | No | true
| Int type narrowing, if true, the tinyint(1) type will be narrowed to
the boolean type if without loss of precision. Support for MySQL at now. Please
refer to `int_type_narrowing` below
[...]
| handle_blob_as_string | Boolean | No | false
| If true, BLOB type will be converted to STRING type. **Only supported
for Oracle database**. This is useful for handling large BLOB fields in Oracle
that exceed the default size limit. When transmitting Oracle's BLOB fields to
systems like Doris, setting this to true can make the data transfer more
efficient.
[...]
+| use_kerberos | Boolean | No | false
| Whether to enable Kerberos authentication for JDBC connections.
[...]
+| kerberos_principal | String | No | -
| Kerberos principal used when `use_kerberos = true`.
[...]
+| kerberos_keytab_path | String | No | -
| Path of the Kerberos keytab file used when `use_kerberos = true`.
[...]
+| krb5_path | String | No |
/etc/krb5.conf | Path of the Kerberos krb5 configuration file.
[...]
| use_select_count | Boolean | No | false
| Use select count for table count rather then other methods in dynamic
chunk split stage. This is currently only available for jdbc-oracle.In this
scenario, select count directly is used when it is faster to update statistics
using sql from analysis table
[...]
| skip_analyze | Boolean | No | false
| Skip the analysis of table count in dynamic chunk split stage. This is
currently only available for jdbc-oracle.In this scenario, you schedule
analysis table sql to update related table statistics periodically or your
table data does not change frequently
[...]
| use_regex | Boolean | No | false
| Control regular expression matching for table_path. When set to `true`,
the table_path will be treated as a regular expression pattern. When set to
`false` or not specified, the table_path will be treated as an exact path (no
regex matching). |
diff --git a/docs/en/connectors/source/MongoDB.md
b/docs/en/connectors/source/MongoDB.md
index 030c572f0f..0cfb58afef 100644
--- a/docs/en/connectors/source/MongoDB.md
+++ b/docs/en/connectors/source/MongoDB.md
@@ -72,7 +72,7 @@ For specific types in MongoDB, we use Extended JSON format to
map them to SeaTun
| 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 | String | Yes | - | MongoDB's
BSON and seatunnel data structure mapping. For more details, please refer to
[Schema Feature](../../introduction/concepts/schema-feature.md).
|
+| 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.
|
@@ -80,7 +80,7 @@ For specific types in MongoDB, we use Extended JSON format to
map them to SeaTun
| 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 | true | By utilizing
flatSyncString, 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.
|
+| 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
|
### Tips
@@ -115,7 +115,7 @@ source {
c_double = double
c_bytes = bytes
c_date = date
- c_decimal = "decimal(38, 18)"
+ c_decimal = "decimal(34, 18)"
c_timestamp = timestamp
c_row = {
c_map = "map<string, string>"
@@ -127,7 +127,7 @@ source {
c_double = double
c_bytes = bytes
c_date = date
- c_decimal = "decimal(38, 18)"
+ c_decimal = "decimal(34, 18)"
c_timestamp = timestamp
}
}
@@ -451,4 +451,3 @@ Use the data samples synchronized with modified parameters,
such as the followin
## Changelog
<ChangeLog />
-
diff --git a/docs/zh/connectors/sink/InfluxDB.md
b/docs/zh/connectors/sink/InfluxDB.md
index b67c53bba4..243bfb38fb 100644
--- a/docs/zh/connectors/sink/InfluxDB.md
+++ b/docs/zh/connectors/sink/InfluxDB.md
@@ -53,6 +53,7 @@ http://influxdb-host:8086
`influxDB` measurement 的名称。这个配置是可选项；不配置时，sink 会使用输入表完整名称作为
measurement 名称，这在多表写入场景很有用。
+多表输入时，请确保生成的表名可以作为合法的 InfluxDB measurement 名称。
### username [string]
@@ -178,6 +179,8 @@ sink {
### 多表写入
+不配置 `measurement` 时，每个上游表会写入一个以该表名命名的 measurement，这是多表输入的常用配置方式。
+
```hocon
env {
parallelism = 1
diff --git a/docs/zh/connectors/sink/Jdbc.md b/docs/zh/connectors/sink/Jdbc.md
index 34feba66dd..5c194e5466 100644
--- a/docs/zh/connectors/sink/Jdbc.md
+++ b/docs/zh/connectors/sink/Jdbc.md
@@ -44,6 +44,8 @@ import ChangeLog from '../changelog/connector-jdbc.md';
| tableSuffix | String | 否 | -
|
| primary_keys | Array | 否 | -
|
| connection_check_timeout_sec | Int | 否 | 30
|
+| connect_timeout_ms | Int | 否 | 86400000
|
+| socket_timeout_ms | Int | 否 | 86400000
|
| max_retries | Int | 否 | 0
|
| batch_size | Int | 否 | 1000
|
| batch_interval_ms | Long | 否 | 0
|
@@ -60,9 +62,16 @@ import ChangeLog from '../changelog/connector-jdbc.md';
| data_save_mode | Enum | 否 | APPEND_DATA
|
| custom_sql | String | 否 | -
|
| enable_upsert | Boolean | 否 | true
|
+| is_primary_key_updated | Boolean | 否 | true
|
+| support_upsert_by_insert_only | Boolean | 否 | false
|
| table_options | Map | 否 | -
|
| use_copy_statement | Boolean | 否 | false
|
| oracle_insert_mode | Enum | 否 | CONVENTIONAL
|
+| create_index | Boolean | 否 | true
|
+| use_kerberos | Boolean | 否 | false
|
+| kerberos_principal | String | 否 | -
|
+| kerberos_keytab_path | String | 否 | -
|
+| krb5_path | String | 否 | /etc/krb5.conf
|
| access_key_id | String | 否 |
|
| secret_access_key | String | 否 |
|
| region | String | 否 |
|
@@ -162,6 +171,14 @@ Tip: 如果目标数据库有 SCHEMA 的概念，则表参数必须写成 `xxx.x
用于验证数据库连接的有效性时等待数据库操作完成所需的时间，单位是秒
+### connect_timeout_ms [int]
+
+建立 JDBC 连接时的连接超时时间，单位毫秒。默认值为 24 小时。设置为 `0` 表示不超时。
+
+### socket_timeout_ms [int]
+
+JDBC 连接建立后的 socket 读取超时时间，单位毫秒。默认值为 24 小时。设置为 `0` 表示不超时。
+
### max_retries [int]
重试提交失败的最大次数（executeBatch）
@@ -277,6 +294,14 @@ sink {
启用通过主键更新插入，如果任务没有key重复数据，设置该参数为 false 可以加快数据导入速度
+### is_primary_key_updated [boolean]
+
+生成 update 语句时是否包含主键字段。除非目标数据库要求更新时跳过主键列，一般保持默认值即可。
+
+### support_upsert_by_insert_only [boolean]
+
+是否通过仅 insert 的语句支持 upsert 行为。该参数属于高级兼容选项，默认关闭。
+
### use_copy_statement [boolean]
使用 `COPY ${table} FROM STDIN` 语句导入数据。仅支持具有 `getCopyAPI()`
方法连接的驱动程序。例如：Postgresql
@@ -296,6 +321,14 @@ INSERT /*+ APPEND_VALUES */ INTO ...
该选项仅支持 Oracle JDBC Sink 的 insert-only 写入。使用时必须配置 `generate_sink_sql =
true`、`auto_commit = true`，不能配置自定义 `query`，不能配置 `primary_keys`，并且
`is_exactly_once = false`、`support_upsert_by_insert_only = false`。
+### create_index [boolean]
+
+自动建表时是否创建索引（包含主键和其他索引）。迁移大表时可关闭该选项以提升写入速度，但迁移完成后需要手动创建索引来保证查询性能。
+
+### use_kerberos [boolean]
+
+是否为 JDBC 连接启用 Kerberos 认证。开启后，请根据环境同时配置
`kerberos_principal`、`kerberos_keytab_path` 和 `krb5_path`。
+
### access_key_id [String]
AWS IAM 认证中所需要的access_key_id 。 该参考仅适用于 dialect="dsql"
diff --git a/docs/zh/connectors/sink/MongoDB.md
b/docs/zh/connectors/sink/MongoDB.md
index eba770ee4a..eb0411e913 100644
--- a/docs/zh/connectors/sink/MongoDB.md
+++ b/docs/zh/connectors/sink/MongoDB.md
@@ -67,16 +67,16 @@ MongoDB 连接器提供从 MongoDB 读取数据以及向 MongoDB 写入数据的
|-----------------------|----------|----------|--------|------|
| 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
集合名称。配置多表同步时，可使用 `${table_name}`、`${schema_name}` 等占位符，例如：`collection =
"${database_name}_${schema_name}_${table_name}_check"`。 |
-| buffer-flush.max-rows | String | 否 | 1000 | 每次批量写入请求的最大缓存行数。 |
-| buffer-flush.interval | String | 否 | 30000 | 批量写入的最大时间间隔（毫秒）。 |
-| retry.max | String | 否 | 3 | 写入失败时的最大重试次数。 |
-| retry.interval | Duration | 否 | 1000 | 写入失败后的重试间隔时间（毫秒）。 |
+| 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 | String | 否 | APPEND_DATA | 数据写入模式：<br/>-
`DROP_DATA`: 插入数据前清空集合；<br/>- `APPEND_DATA`: 追加数据；<br/>-
`ERROR_WHEN_DATA_EXISTS`: 如果集合已有数据则报错。 |
+| data_save_mode | Enum | 否 | APPEND_DATA |
数据写入模式：`DROP_DATA` 表示插入数据前清空集合，`APPEND_DATA` 表示追加数据，`ERROR_WHEN_DATA_EXISTS`
表示集合已有数据时报错。 |
### 提示
@@ -113,13 +113,65 @@ source {
sink {
MongoDB {
- uri = mongodb://user:[email protected]:27017
+ uri = "mongodb://user:[email protected]:27017"
database = "test"
collection = "test"
}
}
```
+### 多表写入
+
+当上游记录带有表元数据时，`database` 和 `collection` 可以使用占位符。常用占位符包括
+`${database_name}`、`${schema_name}` 和 `${table_name}`。
+
+```hocon
+source {
+ FakeSource {
+ tables_configs = [
+ {
+ schema = {
+ table = "testDatabase1.testSchema1.testTable1"
+ fields {
+ id = int
+ value = string
+ }
+ }
+ rows = [
+ {
+ kind = INSERT
+ fields = [1, "NEW"]
+ }
+ ]
+ },
+ {
+ schema = {
+ table = "testDatabase2.testSchema2.testTable2"
+ fields {
+ id = int
+ amount = "decimal(16, 1)"
+ }
+ }
+ rows = [
+ {
+ kind = INSERT
+ fields = [1, 6.3]
+ }
+ ]
+ }
+ ]
+ }
+}
+
+sink {
+ MongoDB {
+ uri = "mongodb://user:[email protected]:27017/test_db?retryWrites=true"
+ database = "test_db"
+ collection = "${database_name}_${schema_name}_${table_name}_check"
+ }
+}
+```
+
## 参数详解
### MongoDB 数据库连接 URI 示例
diff --git a/docs/zh/connectors/source/InfluxDB.md
b/docs/zh/connectors/source/InfluxDB.md
index 46e8e41537..b78b8a4a71 100644
--- a/docs/zh/connectors/source/InfluxDB.md
+++ b/docs/zh/connectors/source/InfluxDB.md
@@ -19,7 +19,7 @@ import ChangeLog from '../changelog/connector-influxdb.md';
支持查询 SQL 并可以实现投影效果。
- [x] [并行性](../../introduction/concepts/connector-v2-features.md)
-- [ ] [支持用户自定义 split](../../introduction/concepts/connector-v2-features.md)
+- [x] [支持用户自定义 split](../../introduction/concepts/connector-v2-features.md)
## 选项
@@ -35,7 +35,7 @@ import ChangeLog from '../changelog/connector-influxdb.md';
| upper_bound | int | 否 | - | 启用并行范围读取时，`split_column` 的上界。
|
| partition_num | int | 否 | 0 | 查询切分数量。`0` 表示不切分，直接执行原始 `sql`。
|
| split_column | string | 否 | - | 分割列
|
-| where | string | 否 | - | 预留的 source 配置项。当前切分逻辑直接从 `sql`
中读取 `where` 关键字。 |
+| where | string | 否 | - | 预留的 source 配置项。当前切分逻辑直接从 `sql`
中读取小写 `where` 关键字。 |
| epoch | string | 否 | n |
返回的时间精度，例如：`H`、`m`、`s`、`MS`、`u`、`n`。 |
| connect_timeout_ms | long | 否 | 15000 | 连接 InfluxDB 的超时时间（毫秒）
|
| query_timeout_sec | int | 否 | 3 | 查询超时时间（秒）
|
@@ -93,6 +93,7 @@ InfluxDB 密码
> - InfluxDB time 不支持作为分割主键，因为 time 字段无法参与数学计算
> - 目前，`split_column` 仅支持整数数据分割，不支持 `float`、`string`、`date` 等类型。
> - `split_column`、`lower_bound`、`upper_bound`、`partition_num` 需要一起配置。
+> - 如果切分读取的 SQL 中包含过滤条件，请在 `sql` 里使用小写 `where`，例如 `select * from test where
age > 0`。当前切分解析逻辑区分大小写。
### upper_bound [int]
diff --git a/docs/zh/connectors/source/Jdbc.md
b/docs/zh/connectors/source/Jdbc.md
index e8fdf2e48b..ebe82210a1 100644
--- a/docs/zh/connectors/source/Jdbc.md
+++ b/docs/zh/connectors/source/Jdbc.md
@@ -51,6 +51,8 @@ import ChangeLog from '../changelog/connector-jdbc.md';
| compatible_mode | String | 否 | - |
数据库的兼容模式，当数据库支持多种兼容模式时需要。<br/> 例如，使用 OceanBase 数据库时，需要将其设置为 'mysql' 或
'oracle'。<br/> 使用 starrocks 时，需要将其设置为 `starrocks`
[...]
| dialect | String | 否 | - |
指定的方言，如果不存在，仍然根据 url 获取，优先级高于 url。<br/> 例如，使用 starrocks 时，需要将其设置为 `starrocks`
[...]
| connection_check_timeout_sec | Int | 否 | 30 |
等待用于验证连接的数据库操作完成的时间（秒）。
[...]
+| connect_timeout_ms | Int | 否 | 86400000 | 建立
JDBC 连接时的连接超时时间，单位毫秒。`0` 表示不超时。
[...]
+| socket_timeout_ms | Int | 否 | 86400000 |
JDBC 连接建立后的 socket 读取超时时间，单位毫秒。`0` 表示不超时。
[...]
| partition_column | String | 否 | - |
用于分割数据的列名。
[...]
| partition_upper_bound | Long | 否 | - |
partition_column 的最大值用于扫描，如果未设置，SeaTunnel 将查询数据库获取最大值。
[...]
| partition_lower_bound | Long | 否 | - |
partition_column 的最小值用于扫描，如果未设置，SeaTunnel 将查询数据库获取最小值。
[...]
@@ -58,6 +60,10 @@ import ChangeLog from '../changelog/connector-jdbc.md';
| decimal_type_narrowing | Boolean | 否 | true |
十进制类型缩小，如果为 true，十进制类型将缩小为 int 或 long 类型（如果没有精度损失）。目前仅支持 Oracle。请参考下面的
`decimal_type_narrowing`
[...]
| int_type_narrowing | Boolean | 否 | true | Int
类型缩小，如果为 true，tinyint(1) 类型将缩小为布尔类型（如果没有精度损失）。目前支持 MySQL。请参考下面的
`int_type_narrowing`
[...]
| handle_blob_as_string | Boolean | 否 | false | 如果为
true，BLOB 类型将转换为 STRING 类型。**仅支持 Oracle 数据库**。这对于处理超过默认大小限制的 Oracle 中的大 BLOB
字段很有用。将 Oracle 的 BLOB 字段传输到 Doris 等系统时，将其设置为 true 可以使数据传输更高效。
[...]
+| use_kerberos | Boolean | 否 | false | 是否为
JDBC 连接启用 Kerberos 认证。
[...]
+| kerberos_principal | String | 否 | - |
`use_kerberos = true` 时使用的 Kerberos principal。
[...]
+| kerberos_keytab_path | String | 否 | - |
`use_kerberos = true` 时使用的 Kerberos keytab 文件路径。
[...]
+| krb5_path | String | 否 | /etc/krb5.conf
| Kerberos krb5 配置文件路径。
[...]
| use_select_count | Boolean | 否 | false |
在动态块分割阶段使用 select count 来获取表计数，而不是其他方法。这目前仅适用于 jdbc-oracle。在这种情况下，当使用 sql
从分析表更新统计信息更快时，直接使用 select count
[...]
| skip_analyze | Boolean | 否 | false |
在动态块分割阶段跳过表计数分析。这目前仅适用于 jdbc-oracle。在这种情况下，您定期安排分析表 sql 来更新相关表统计信息，或您的表数据不经常更改
[...]
| use_regex | Boolean | 否 | false | 控制
table_path 的正则表达式匹配。设置为 `true` 时，table_path 将被视为正则表达式模式。设置为 `false`
或未指定时，table_path 将被视为精确路径（无正则表达式匹配）。 |
diff --git a/docs/zh/connectors/source/MongoDB.md
b/docs/zh/connectors/source/MongoDB.md
index e6c695240c..490dd694a6 100644
--- a/docs/zh/connectors/source/MongoDB.md
+++ b/docs/zh/connectors/source/MongoDB.md
@@ -72,7 +72,7 @@ MongoDB连接器提供了从MongoDB读取数据和向MongoDB写入数据的能
| uri | String | 是 | - | MongoDB标准连接uri。例如
mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true.
|
| database | String | 是 | - |
要读取或写入的MongoDB数据库的名称。
|
| collection | String | 是 | - |
要读取或写入的MongoDB集合的名称。
|
-| schema | String | 是 | - |
MongoDB的BSON和seatunnel数据结构映射。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
|
+| schema | Config | 是 | - | MongoDB 的 BSON 与
SeaTunnel 数据结构映射。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
|
| match.query | String | 否 | - |
在MongoDB中，过滤器用于过滤查询操作的文档。
|
| match.projection | String | 否 | - |
在MongoDB中，投影用于控制查询结果中包含的字段。
|
| partition.split-key | String | 否 | _id | 分片字段。
|
@@ -80,7 +80,7 @@ MongoDB连接器提供了从MongoDB读取数据和向MongoDB写入数据的能
| 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 | 否 | true |
通过使用flatSyncString，只能设置一个字段属性值，字段类型必须是String。此操作将对单个MongoDB数据条目执行字符串映射。
|
+| flat.sync-string | Boolean | 否 | false | 开启后，会把整个 MongoDB
文档映射到一个 SeaTunnel `STRING` 字段。此时 schema 只能配置一个字段，并且该字段类型必须是 `STRING`。
|
| common-options | | 否 | - | 源插件常用参数，请参考
[源通用选项](../common-options/source-common-options.md)
|
### 提示
@@ -115,7 +115,7 @@ source {
c_double = double
c_bytes = bytes
c_date = date
- c_decimal = "decimal(38, 18)"
+ c_decimal = "decimal(34, 18)"
c_timestamp = timestamp
c_row = {
c_map = "map<string, string>"
@@ -127,7 +127,7 @@ source {
c_double = double
c_bytes = bytes
c_date = date
- c_decimal = "decimal(38, 18)"
+ c_decimal = "decimal(34, 18)"
c_timestamp = timestamp
}
}
@@ -451,4 +451,3 @@ sink {
## 修改日志
<ChangeLog />
-
