This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 582e47e671 [Docs][Connector-V2] Improve Greenplum Mqtt and SqlServer
connector docs (#11798)
582e47e671 is described below
commit 582e47e671b5b9fa83a570cd820e467de36373c6
Author: Daniel Carter <[email protected]>
AuthorDate: Fri Aug 14 17:02:56 2026 +0800
[Docs][Connector-V2] Improve Greenplum Mqtt and SqlServer connector docs
(#11798)
Co-authored-by: DanielCarter-stack
<[email protected]>
---
docs/zh/connectors/sink/Greenplum.md | 65 ++++++++-
docs/zh/connectors/source/Mqtt.md | 183 +++++++++++-------------
docs/zh/connectors/source/SqlServer.md | 253 ++++++++++++++++++---------------
3 files changed, 289 insertions(+), 212 deletions(-)
diff --git a/docs/zh/connectors/sink/Greenplum.md
b/docs/zh/connectors/sink/Greenplum.md
index 3e99f9ee21..3d82083c9e 100644
--- a/docs/zh/connectors/sink/Greenplum.md
+++ b/docs/zh/connectors/sink/Greenplum.md
@@ -26,10 +26,11 @@ import ChangeLog from '../changelog/connector-jdbc.md';
> 1. 使用 `org.postgresql.Driver` 时，请确保 [PostgreSQL JDBC
> 驱动](https://mvnrepository.com/artifact/org.postgresql/postgresql) 已放到
> `${SEATUNNEL_HOME}/lib/`。
> 2. 使用 `com.pivotal.jdbc.GreenplumDriver` 时，请自行下载 Greenplum JDBC 驱动，并放到
> `${SEATUNNEL_HOME}/lib/`。
-## 关键特性
+## 主要特性
- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md)
- [ ] [变更数据捕获](../../introduction/concepts/connector-v2-features.md)
+- [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md)
- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md)
:::tip
@@ -45,9 +46,28 @@ Greenplum Sink 不支持精确一次语义，因为 Greenplum 不支持 XA 事
| 使用 PostgreSQL 驱动连接 Greenplum | `org.postgresql.Driver` |
`jdbc:postgresql://localhost:5432/testdb` |
[下载](https://mvnrepository.com/artifact/org.postgresql/postgresql) |
| 使用 Greenplum 原生驱动连接 Greenplum | `com.pivotal.jdbc.GreenplumDriver` |
`jdbc:pivotal:greenplum://localhost:5432;DatabaseName=testdb` | 从 Greenplum 获取 |
+## 数据类型映射
+
+Greenplum 沿用 PostgreSQL JDBC 驱动映射。下表列出常用类型对照：
+
+| Greenplum 数据类型 | SeaTunnel 数据类型 |
+|--------------------|--------------------|
+| BOOLEAN | BOOLEAN |
+| SMALLINT / INT2 | SMALLINT |
+| INT / INT4 / SERIAL | INT |
+| BIGINT / INT8 / BIGSERIAL | BIGINT |
+| NUMERIC(p, s) / DECIMAL(p, s) / MONEY | DECIMAL(p, s) |
+| REAL / FLOAT4 | FLOAT |
+| DOUBLE PRECISION / FLOAT8 | DOUBLE |
+| CHAR / VARCHAR / TEXT / JSON / JSONB | STRING |
+| DATE | DATE |
+| TIME | TIME |
+| TIMESTAMP / TIMESTAMPTZ | TIMESTAMP |
+| BYTEA | BYTES |
+
## 选项
-这里只列出 Greenplum 常用配置。其他 JDBC Sink 配置，例如
`batch_size`、`max_retries`、`generate_sink_sql`、`database`、`table`、`primary_keys`
和 `properties`，继承自 [Jdbc Sink](Jdbc.md)。
+这里只列出 Greenplum 常用配置。其他 JDBC Sink 配置，例如
`batch_size`、`max_retries`、`generate_sink_sql`、`database`、`table`、`primary_keys`、`connection_check_timeout_sec`、`max_commit_attempts`
等，继承自 [Jdbc Sink](Jdbc.md)。
| 名称 | 类型 | 是否必填 | 默认值 | 描述 |
|------|------|----------|--------|------|
@@ -61,12 +81,19 @@ Greenplum Sink 不支持精确一次语义，因为 Greenplum 不支持 XA 事
| generate_sink_sql | Boolean | 否 | false | 是否根据 `database` 和 `table` 自动生成插入
SQL。 |
| database | String | 否 | - | `generate_sink_sql = true` 时使用的数据库名。 |
| table | String | 否 | - | `generate_sink_sql = true` 时使用的目标表名。 |
+| primary_keys | Array | 否 | - | 自动生成 SQL 时用于 upsert 语义的主键字段列表。 |
+| connection_check_timeout_sec | Int | 否 | 30 | 验证数据库连接操作的超时时间（秒）。 |
+| max_commit_attempts | Int | 否 | 3 | 事务提交失败时的最大重试次数。 |
+| transaction_timeout_sec | Int | 否 | -1 | 事务超时时间（秒）。`-1` 表示无限制。 |
+| enable_upsert | Boolean | 否 | true | 是否启用基于主键的 upsert 写入。 |
| common-options | | 否 | - | Sink 插件通用参数，请参考 [Sink
通用选项](../common-options/sink-common-options.md)。 |
:::tip
出于许可证原因，SeaTunnel 不内置 Greenplum 原生 JDBC 驱动。如果使用
`com.pivotal.jdbc.GreenplumDriver`，请在运行任务前将 `greenplum-xxx.jar` 复制到对应引擎的依赖目录。
+Greenplum 不支持 XA 事务，因此 Sink 端无法保证精确一次语义。如果你需要端到端一致，请结合外部存储（例如
Kafka、Hudi）使用幂等批写。
+
:::
## 任务示例
@@ -149,6 +176,40 @@ sink {
}
```
+### CDC 数据流写入
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 10000
+}
+
+source {
+ MySQL-CDC {
+ username = "cdc_user"
+ password = "cdc_pass"
+ table-list = ["cdc_test.orders"]
+ base-url = "jdbc:mysql://localhost:3306/cdc_test"
+ startup.mode = "initial"
+ }
+}
+
+sink {
+ Jdbc {
+ driver = "org.postgresql.Driver"
+ url = "jdbc:postgresql://localhost:5432/testdb"
+ username = "tester"
+ password = "pivotal"
+ generate_sink_sql = true
+ database = "testdb"
+ table = "orders"
+ primary_keys = ["id"]
+ enable_upsert = true
+ }
+}
+```
+
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/source/Mqtt.md
b/docs/zh/connectors/source/Mqtt.md
index f6624aa725..b3b129a9e6 100644
--- a/docs/zh/connectors/source/Mqtt.md
+++ b/docs/zh/connectors/source/Mqtt.md
@@ -37,104 +37,29 @@ import ChangeLog from '../changelog/connector-mqtt.md';
## 选项
-| 参数名 | 类型 | 必须 | 默认值 |
-|-------------------|---------|----|------|
-| url | string | 是 | - |
-| topic | string | 是 | - |
-| schema | config | 是 | - |
-| username | string | 否 | - |
-| password | string | 否 | - |
-| qos | int | 否 | 1 |
-| format | string | 否 | json |
-| field_delimiter | string | 否 | , |
-| client_id | string | 否 | - |
-| clean_session | boolean | 否 | true |
-| connection_timeout | int | 否 | 30 |
-| keep_alive_interval | int | 否 | 60 |
-| reconnect_timeout | int | 否 | 120 |
-| max_queue_size | int | 否 | 1000 |
-| common-options | | 否 | - |
+| 参数名 | 类型 | 是否必填 | 默认值 | 描述 |
+|--------|------|----------|--------|------|
+| url | String | 是 | - | MQTT broker 连接 URL，必须包含协议、主机和端口，例如
`tcp://broker.example.com:1883`。MQTT over TLS/SSL 请使用
`ssl://broker.example.com:8883`。 |
+| topic | String | 是 | - | 要订阅消息的 MQTT topic，例如 `iot/sensors/temperature`。 |
+| schema | Config | 是 | - | 上游数据的 schema 字段，详见 [Schema
特性](../../introduction/concepts/schema-feature.md)。 |
+| username | String | 否 | - | MQTT broker 认证用户名，匿名访问时可不填。 |
+| password | String | 否 | - | MQTT broker 认证密码，匿名访问时可不填。 |
+| qos | Int | 否 | 1 | 订阅 topic 时使用的 MQTT QoS 等级。支持 `0`（QoS 0）或 `1`（QoS
1）。该设置只控制 MQTT broker 和 client 之间的交付，不提供 SeaTunnel 端到端保证。 |
+| format | String | 否 | json | 输入消息的反序列化格式。支持 `json`（将每条消息反序列化为 JSON 对象）或
`text`（按 `field_delimiter` 切分为纯文本）。 |
+| field_delimiter | String | 否 | `,` | 当 `format=text` 时使用的字段分隔符，例如
`,`、`\|`、`\t`。 |
+| client_id | String | 否 | - | MQTT client id。当 `clean_session=true`
且未配置该选项时，连接器会生成随机 client id。`clean_session=false` 时必须配置稳定的 `client_id`。 |
+| clean_session | Boolean | 否 | true | 是否使用 clean MQTT session。`true` 时 broker
丢弃之前的会话状态；`false` 时 broker 可以保留会话状态和订阅，需要稳定的 `client_id`。 |
+| connection_timeout | Int | 否 | 30 | MQTT 连接建立超时时间，单位为秒。 |
+| keep_alive_interval | Int | 否 | 60 | MQTT keep alive 间隔，单位为秒。 |
+| reconnect_timeout | Int | 否 | 120 | 等待 MQTT 自动重连的最长时间，单位为秒。如果 MQTT
客户端断开连接的时间超过该超时时间，源任务会失败，避免无限期静默等待。 |
+| max_queue_size | Int | 否 | 1000 | 反序列化之前在内存中缓存的 MQTT 消息最大数量。 |
+| common-options | | 否 | - | 源插件通用参数，详情请参考
[源通用选项](../common-options/source-common-options.md)。 |
+
+:::tip
+
+该连接器当前只支持单个 topic 订阅，多 topic 拆分请使用多个 MQTT source。MQTT 5.0 特性（如共享订阅、消息属性）暂未启用。
-### url [string]
-
-MQTT broker 连接 URL。必须包含协议、主机和端口。
-
-示例：`tcp://broker.example.com:1883`
-
-### topic [string]
-
-要订阅消息的 MQTT topic。
-
-示例：`iot/sensors/temperature`
-
-### schema [config]
-
-上游数据的 schema 字段。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
-
-### username [string]
-
-MQTT broker 认证用户名。匿名访问时可以不配置。
-
-### password [string]
-
-MQTT broker 认证密码。匿名访问时可以不配置。
-
-### qos [int]
-
-订阅 topic 时使用的 MQTT Quality of Service 等级。
-
-该设置只控制 MQTT broker 和 MQTT client 之间的交付，不提供 SeaTunnel 端到端交付保证。
-
-支持的值：
-
-- `0` — MQTT QoS 0
-- `1` — MQTT QoS 1
-
-### format [string]
-
-输入消息的反序列化格式。支持的值：
-
-- `json` — 将每条消息反序列化为 JSON 对象（默认）
-- `text` — 将每条消息按分隔符反序列化为纯文本（分隔符由 `field_delimiter` 控制）
-
-### field_delimiter [string]
-
-当 `format` 设置为 `text` 时使用的字段分隔符。默认值为 `,`。
-
-示例：`,`, `|`, `\t`
-
-### client_id [string]
-
-MQTT client id。当 `clean_session=true` 且未配置该选项时，连接器会生成随机 client id。
-
-当 `clean_session=false` 时必须配置该选项，因为 MQTT 持久会话需要稳定的 client id。
-
-### clean_session [boolean]
-
-是否使用 clean MQTT session。默认值为 `true`。
-
-- `true` — broker 丢弃之前的会话状态，适合无状态运行。
-- `false` — broker 可以保留会话状态，包括订阅信息。需要稳定的 `client_id`。
-
-### connection_timeout [int]
-
-MQTT 连接建立超时时间，单位为秒。
-
-### keep_alive_interval [int]
-
-MQTT keep alive 间隔，单位为秒。
-
-### reconnect_timeout [int]
-
-等待 MQTT 自动重连的最长时间，单位为秒。如果 MQTT 客户端断开连接的时间超过该超时时间，`pollNext()`
会使源任务失败，避免无限期静默等待。
-
-### max_queue_size [int]
-
-反序列化之前在内存中缓存的 MQTT 消息最大数量。
-
-### common options
-
-源插件通用参数，详情请参考 [源通用选项](../common-options/source-common-options.md)。
+:::
## 示例
@@ -200,6 +125,70 @@ sink {
}
```
+### TLS/SSL 接入
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ MQTT {
+ url = "ssl://broker.example.com:8883"
+ topic = "factory/line-1/status"
+ username = "seatunnel"
+ password = "broker-token"
+ client_id = "seatunnel-mqtt-tls"
+ qos = 1
+ connection_timeout = 30
+ keep_alive_interval = 60
+ reconnect_timeout = 180
+ format = "json"
+ schema = {
+ fields {
+ device_id = string
+ status = string
+ ts = bigint
+ }
+ }
+ }
+}
+
+sink {
+ Console {}
+}
+```
+
+### 文本分隔源
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ MQTT {
+ url = "tcp://broker.example.com:1883"
+ topic = "factory/line-2/log"
+ format = "text"
+ field_delimiter = "|"
+ schema = {
+ fields {
+ device_id = string
+ level = string
+ message = string
+ }
+ }
+ }
+}
+
+sink {
+ Console {}
+}
+```
+
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/source/SqlServer.md
b/docs/zh/connectors/source/SqlServer.md
index e7c781b95d..6c13959aec 100644
--- a/docs/zh/connectors/source/SqlServer.md
+++ b/docs/zh/connectors/source/SqlServer.md
@@ -38,105 +38,106 @@ import ChangeLog from '../changelog/connector-jdbc.md';
## 描述
-通过 JDBC 读取外部数据源数据。
+通过 JDBC 读取外部数据源数据。SQL Server 的更多通用选项请参考 [Jdbc 连接器](Jdbc.md)。
## 支持的数据源信息
-| 数据源 | 支持版本 | 驱动 |
url | maven
|
-|------------|-------------------------|----------------------------------------------|---------------------------------|-----------------------------------------------------------------------------------|
-| SQL Server | 支持版本 >= 2008 |
com.microsoft.sqlserver.jdbc.SQLServerDriver | jdbc:sqlserver://localhost:1433
| [下载](https://mvnrepository.com/artifact/com.microsoft.sqlserver/mssql-jdbc) |
+| 数据源 | 支持版本 | 驱动 | url | maven |
+|--------|----------|------|-----|-------|
+| SQL Server | 支持版本 >= 2008 | `com.microsoft.sqlserver.jdbc.SQLServerDriver` |
`jdbc:sqlserver://localhost:1433` |
[下载](https://mvnrepository.com/artifact/com.microsoft.sqlserver/mssql-jdbc) |
## 数据库依赖
-> 请下载对应 'Maven' 的支持列表，并将其复制到 '$SEATUNNEL_HOME/plugins/jdbc/lib/' 工作目录<br/>
-> 例如 SQL Server 数据源：cp mssql-jdbc-xxx.jar $SEATUNNEL_HOME/plugins/jdbc/lib/
+> 请下载对应 'Maven' 的支持列表，并将其复制到 `$SEATUNNEL_HOME/plugins/jdbc/lib/` 工作目录。<br/>
+> 例如 SQL Server 数据源：`cp mssql-jdbc-xxx.jar $SEATUNNEL_HOME/plugins/jdbc/lib/`
## 数据类型映射
-| SQLserver 数据类型 | Seatunnel
数据类型 |
-|----------------------------------------------------------------------|---------------------|
-| BIT |
BOOLEAN |
-| TINYINT<br/>SMALLINT |
SMALLINT |
-| INTEGER<br/>INT | INT
|
-| BIGINT |
BIGINT |
-| NUMERIC(p,s)<br/>DECIMAL(p,s)<br/>MONEY<br/>SMALLMONEY |
DECIMAL(p,s) |
-| FLOAT(1~24)<br/>REAL | FLOAT
|
-| DOUBLE<br/>FLOAT(>24) |
DOUBLE |
-| CHAR<br/>NCHAR<br/>VARCHAR<br/>NTEXT<br/>NVARCHAR<br/>TEXT<br/>XML |
STRING |
-| DATE | DATE
|
-| TIME(s) |
TIME(s) |
-| DATETIME(s)<br/>DATETIME2(s)<br/>DATETIMEOFFSET(s)<br/>SMALLDATETIME |
TIMESTAMP(s) |
-| BINARY<br/>VARBINARY<br/>IMAGE | BYTES
|
+| SQL Server 数据类型 | SeaTunnel 数据类型 |
+|---------------------|---------------------|
+| BIT | BOOLEAN |
+| TINYINT / SMALLINT | SMALLINT |
+| INTEGER / INT | INT |
+| BIGINT | BIGINT |
+| NUMERIC(p,s) / DECIMAL(p,s) / MONEY / SMALLMONEY | DECIMAL(p,s) |
+| FLOAT(1~24) / REAL | FLOAT |
+| DOUBLE / FLOAT(>24) | DOUBLE |
+| CHAR / NCHAR / VARCHAR / NTEXT / NVARCHAR / TEXT / XML | STRING |
+| DATE | DATE |
+| TIME(s) | TIME(s) |
+| DATETIME / DATETIME2 / DATETIMEOFFSET / SMALLDATETIME | TIMESTAMP(s) |
+| BINARY / VARBINARY / IMAGE | BYTES |
## 数据源参数
-| 名称 | 类型 | 是否必填 | 默认值 | 描述
|
-| ------------------------------------------ | ------- | -------- |
--------------- |
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
-| url | String | 是 | -
| JDBC 连接的 URL。参见示例：jdbc:sqlserver://127.0.0.1:1434;database=TestDB
|
-| driver | String | 是 | -
| 用于连接远程数据源的 jdbc 类名，<br/>如果使用 SQLserver，值为
`com.microsoft.sqlserver.jdbc.SQLServerDriver`。
|
-| username | String | 否 | -
| 连接实例的用户名
|
-| password | String | 否 | -
| 连接实例的密码
|
-| query | String | 否 | -
| 查询语句。当未配置 `table_path` 和 `table_list` 时必填。
|
-| connection_check_timeout_sec | Int | 否 | 30
| 等待用于验证连接的数据库操作完成的时间（秒）
|
-| partition_column | String | 否 | -
| 用于并行度分区的列名，仅支持数值类型。
|
-| partition_lower_bound | Long | 否 | -
| partition_column 扫描的最小值，如果未设置，SeaTunnel 将查询数据库获取最小值。
|
-| partition_upper_bound | Long | 否 | -
| partition_column 扫描的最大值，如果未设置，SeaTunnel 将查询数据库获取最大值。
|
-| partition_num | Int | 否 | job
parallelism | 分区数量，仅支持正整数。默认值为作业并行度
|
-| fetch_size | Int | 否 | 0
|
对于返回大量对象的查询，你可以配置<br/>查询中使用的行获取大小来提高性能，<br/>通过减少满足选择条件所需的数据库命中次数。<br/>零表示使用
jdbc 默认值。
|
-| properties | Map | 否 | -
| 额外的连接配置参数，当 properties 和 URL 具有相同参数时，优先级由<br/>驱动的具体实现决定。例如，在 MySQL
中，properties 优先于 URL。
|
-| use_regex | Boolean | 否 | false
| 控制 table_path 的正则表达式匹配。当设置为 `true` 时，table_path 将被视为正则表达式模式。当设置为 `false`
或未指定时，table_path 将被视为精确路径（不进行正则匹配）。
|
-| table_path | String | 否 | -
| 表的完整路径，您可以使用此配置代替 `query`。<br/>示例：<br/>"testdb.test_schema.table1"
|
-| table_list | Array | 否 | -
| 要读取的表列表，您可以使用此配置代替 `table_path`。示例：```[{ table_path = "testdb.table1"},
{table_path = "testdb.table2", query = "select id, name from
testdb.table2"}]```
|
-| where_condition | String | 否 | -
| 所有表/查询的通用行过滤条件，必须以 `where` 开头。例如 `where id > 100`
|
-| split.size | Int | 否 | 8096
| 表的分割大小（行数），读取表时，捕获的表会被分割为多个分割。
|
-| split.even-distribution.factor.lower-bound | Double | 否 | 0.05
| 分块键分布因子的下界。此因子用于确定表数据是否均匀分布。如果计算的分布因子大于或等于此下界（即，(MAX(id) - MIN(id) + 1) /
行数），表分块将被优化以实现均匀分布。否则，如果分布因子较小，如果估计的分片数超过 `sample-sharding.threshold`
指定的值，表将被视为不均匀分布并使用基于采样的分片策略。默认值为 0.05。 |
-| split.even-distribution.factor.upper-bound | Double | 否 | 100
| 分块键分布因子的上界。此因子用于确定表数据是否均匀分布。如果计算的分布因子小于或等于此上界（即，(MAX(id) - MIN(id) + 1) /
行数），表分块将被优化以实现均匀分布。否则，如果分布因子较大，如果估计的分片数超过 `sample-sharding.threshold`
指定的值，表将被视为不均匀分布并使用基于采样的分片策略。默认值为 100.0。 |
-| split.sample-sharding.threshold | Int | 否 | 1000
| 此配置指定了触发采样分片策略的估计分片数阈值。当分布因子超出
`chunk-key.even-distribution.factor.upper-bound` 和
`chunk-key.even-distribution.factor.lower-bound` 指定的范围，并且估计的分片数（计算为近似行数 /
分块大小）超过此阈值时，将使用采样分片策略。这可以帮助更有效地处理大型数据集。默认值为 1000 分片。 |
-| split.allow-sampling | Boolean | 否 | true
| 是否允许对分布不均匀的分片键使用采样分片策略。设置为 `false` 时，SeaTunnel 会退回到迭代式不均匀分片。
|
-| use_select_count | Boolean | 否 | false
| 是否使用 `select count(*)` 在分片前估算表行数。
|
-| skip_analyze | Boolean | 否 | false
| 是否跳过分片前的表行数分析。
|
-| split.inverse-sampling.rate | Int | 否 | 1000
| 采样分片策略中使用的采样率的倒数。例如，如果此值设置为 1000，则意味着在采样过程中应用 1/1000
的采样率。此选项提供了控制采样粒度的灵活性，从而影响最终的分片数量。对于非常大的数据集，首选较低的采样率时，此选项特别有用。默认值为 1000。
|
-| common-options | | 否 | -
| 源插件通用参数，请参考 [源通用选项](../common-options/source-common-options.md) 获取详细信息
|
+| 名称 | 类型 | 是否必填 | 默认值 | 描述 |
+|------|------|----------|--------|------|
+| url | String | 是 | - | JDBC 连接的 URL，例如
`jdbc:sqlserver://127.0.0.1:1434;database=TestDB`。 |
+| driver | String | 是 | - | 用于连接远程数据源的 JDBC 类名，SQL Server 使用
`com.microsoft.sqlserver.jdbc.SQLServerDriver`。 |
+| username | String | 否 | - | 连接实例的用户名。 |
+| password | String | 否 | - | 连接实例的密码。 |
+| query | String | 否 | - | 查询语句。当未配置 `table_path` 和 `table_list` 时必填。 |
+| connection_check_timeout_sec | Int | 否 | 30 | 等待用于验证连接的数据库操作完成的时间（秒）。 |
+| partition_column | String | 否 | - | 用于并行度分区的列名，仅支持数值类型。 |
+| partition_lower_bound | Long | 否 | - | `partition_column`
扫描的最小值，如果未设置，SeaTunnel 将查询数据库获取最小值。 |
+| partition_upper_bound | Long | 否 | - | `partition_column`
扫描的最大值，如果未设置，SeaTunnel 将查询数据库获取最大值。 |
+| partition_num | Int | 否 | job parallelism | 分区数量，仅支持正整数。默认值为作业并行度。 |
+| fetch_size | Int | 否 | 0 | 查询使用的行获取大小。`0` 表示使用 JDBC 默认值。增大可减少对数据库的命中次数。 |
+| properties | Map | 否 | - | 额外的连接配置参数。当 properties 与 URL
含相同参数时，由驱动的具体实现决定优先级。 |
+| use_regex | Boolean | 否 | false | 控制 `table_path` 的正则匹配。`true`
时按正则匹配，`false`（默认）时按精确路径匹配。 |
+| table_path | String | 否 | - | 表的完整路径，可用于替代
`query`。示例：`testdb.test_schema.table1`。 |
+| table_list | Array | 否 | - | 要读取的表列表，可替代 `table_path`。示例：`[{ table_path =
"testdb.table1"}, {table_path = "testdb.table2", query = "select id, name from
testdb.table2"}]`。 |
+| where_condition | String | 否 | - | 所有表/查询的通用行过滤条件，必须以 `where` 开头。例如 `where
id > 100`。 |
+| split.size | Int | 否 | 8096 | 表的分割大小（行数），读取表时会被分割为多个 split。 |
+| split.even-distribution.factor.lower-bound | Double | 否 | 0.05 |
分块键分布因子的下界。`(MAX(id) - MIN(id) + 1) / 行数` 大于等于该下界时认为表分布均匀；否则进入采样分片路径。 |
+| split.even-distribution.factor.upper-bound | Double | 否 | 100 |
分块键分布因子的上界。上界取值的逻辑与下界对称。 |
+| split.sample-sharding.threshold | Int | 否 | 1000 | 触发采样分片策略的估算分片数阈值。 |
+| split.allow-sampling | Boolean | 否 | true | 是否允许对分布不均匀的分片键使用采样分片策略。`false`
时退回到迭代式不均匀分片。 |
+| use_select_count | Boolean | 否 | false | 是否在分片前用 `select count(*)` 估算表行数。 |
+| skip_analyze | Boolean | 否 | false | 是否跳过分片前的表行数分析。 |
+| split.inverse-sampling.rate | Int | 否 | 1000 | 采样分片策略中采样率的倒数。`1000` 表示
1/1000 的采样率。 |
+| common-options | | 否 | - | 源插件通用参数，请参考
[源通用选项](../common-options/source-common-options.md)。 |
## 并行读取器
-JDBC 源连接器支持从表中并行读取数据。SeaTunnel 将使用某些规则来分割表中的数据，然后将其交给读取器进行读取。读取器的数量由
`parallelism` 选项决定。
+JDBC 源连接器支持从表中并行读取数据。SeaTunnel 将使用某些规则来分割表中的数据，然后交给读取器进行读取。读取器的数量由
`parallelism` 选项决定。
**分割键规则：**
1. 如果 `partition_column` 不为空，将使用它来计算分割。该列必须在 **支持的分割数据类型** 中。
-2. 如果 `partition_column` 为空，seatunnel 将从表中读取模式并获取主键和唯一索引。如果主键和唯一索引中有多个列，则将使用
**支持的分割数据类型** 中的第一列来分割数据。例如，表具有主键(nn guid, name varchar)，因为 `guid` 不在
**支持的分割数据类型** 中，所以将使用 `name` 列来分割数据。
+2. 如果 `partition_column` 为空，SeaTunnel 将从表中读取模式并获取主键和唯一索引。如果主键和唯一索引中有多个列，则将使用
**支持的分割数据类型** 中的第一列来分割数据。
**支持的分割数据类型：**
+
* String
-* Number(int, bigint, decimal, ...)
+* Number（int、bigint、decimal 等）
* Date
-### 与分割相关的选项
+### 分割相关选项
#### split.size
-一个分割中有多少行，读取表时，捕获的表会被分割为多个分割。
+一个 split 中包含多少行。表被读入时会被分割为多个 split。
#### split.even-distribution.factor.lower-bound
> 不推荐使用
-分块键分布因子的下界。此因子用于确定表数据是否均匀分布。如果计算的分布因子大于或等于此下界（即，(MAX(id) - MIN(id) + 1) /
行数），表分块将被优化以实现均匀分布。否则，如果分布因子较小，如果估计的分片数超过 `sample-sharding.threshold`
指定的值，表将被视为不均匀分布并使用基于采样的分片策略。默认值为 0.05。
+分块键分布因子的下界。`(MAX(id) - MIN(id) + 1) / 行数` 大于等于该下界时认为表分布均匀；否则进入采样分片路径。默认值为 0.05。
#### split.even-distribution.factor.upper-bound
> 不推荐使用
-分块键分布因子的上界。此因子用于确定表数据是否均匀分布。如果计算的分布因子小于或等于此上界（即，(MAX(id) - MIN(id) + 1) /
行数），表分块将被优化以实现均匀分布。否则，如果分布因子较大，如果估计的分片数超过 `sample-sharding.threshold`
指定的值，表将被视为不均匀分布并使用基于采样的分片策略。默认值为 100.0。
+分块键分布因子的上界。逻辑与下界对称。默认值为 100.0。
#### split.sample-sharding.threshold
-此配置指定了触发采样分片策略的估计分片数阈值。当分布因子超出
`chunk-key.even-distribution.factor.upper-bound` 和
`chunk-key.even-distribution.factor.lower-bound` 指定的范围，并且估计的分片数（计算为近似行数 /
分块大小）超过此阈值时，将使用采样分片策略。这可以帮助更有效地处理大型数据集。默认值为 1000 分片。
+触发采样分片策略的估算分片数阈值。默认值为 1000。
#### split.inverse-sampling.rate
-采样分片策略中使用的采样率的倒数。例如，如果此值设置为 1000，则意味着在采样过程中应用 1/1000
的采样率。此选项提供了控制采样粒度的灵活性，从而影响最终的分片数量。对于非常大的数据集，首选较低的采样率时，此选项特别有用。默认值为 1000。
+采样分片策略中采样率的倒数。默认值为 1000（1/1000 的采样率）。
#### partition_column [string]
@@ -144,23 +145,23 @@ JDBC 源连接器支持从表中并行读取数据。SeaTunnel 将使用某些
#### partition_upper_bound [BigDecimal]
-partition_column 扫描的最大值，如果未设置，SeaTunnel 将查询数据库获取最大值。
+`partition_column` 扫描的最大值。如果未设置，SeaTunnel 将查询数据库获取最大值。
#### partition_lower_bound [BigDecimal]
-partition_column 扫描的最小值，如果未设置，SeaTunnel 将查询数据库获取最小值。
+`partition_column` 扫描的最小值。如果未设置，SeaTunnel 将查询数据库获取最小值。
#### partition_num [int]
> 不推荐使用，正确的方法是通过 `split.size` 控制分割数量
-我们需要分割为多少个分割，仅支持正整数。默认值为作业并行度。
+需要分割为多少个 split，仅支持正整数。默认值为作业并行度。
## 提示
-> 如果表无法分割（例如，表没有主键或唯一索引，且未设置 `partition_column`），将以单个并发运行。
+> 如果表无法分割（例如，表没有主键或唯一索引，且未设置 `partition_column`），将以单并发运行。
>
-> 使用 `table_path` 替代 `query` 进行单表读取。如果需要读取多个表，请使用 `table_list`。
+> 单表读取可使用 `table_path` 替代 `query`，多表读取请使用 `table_list`。
## 任务示例
@@ -168,106 +169,132 @@ partition_column 扫描的最小值，如果未设置，SeaTunnel 将查询数
> 读取数据表的简单单个任务
-```
-# 定义运行时环境
+```hocon
env {
parallelism = 1
job.mode = "BATCH"
}
-source{
- Jdbc {
- driver = com.microsoft.sqlserver.jdbc.SQLServerDriver
- url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
- username = SA
- password = "Y.sa123456"
- query = "select * from full_types_jdbc"
- }
+
+source {
+ Jdbc {
+ driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
+ url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
+ username = "SA"
+ password = "Y.sa123456"
+ query = "select * from full_types_jdbc"
+ }
}
transform {
- # 如果你想了解更多关于如何配置 seatunnel 的信息，并查看转换插件的完整列表，
- # 请前往 https://seatunnel.apache.org/docs/transforms/sql
+ # 如果你想了解更多关于如何配置 seatunnel 的信息，并查看转换插件的完整列表，
+ # 请前往 https://seatunnel.apache.org/docs/transforms/sql
}
sink {
- Console {}
+ Console {}
}
```
### 并行示例
-> 使用您配置的分片字段并行读取查询表和分片数据。如果您想读取整个表，可以这样做
+> 使用您配置的分区字段并行读取数据。如果需要读取整张表，可以结合 `query` 或 `table_path` 使用。
-```
+```hocon
env {
parallelism = 10
job.mode = "BATCH"
}
source {
- Jdbc {
- driver = com.microsoft.sqlserver.jdbc.SQLServerDriver
- url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
- username = SA
- password = "Y.sa123456"
- # 根据需要定义查询逻辑
- query = "select * from full_types_jdbc"
- # 并行分片读取字段
- partition_column = "id"
- # 分片数量
- partition_num = 10
- }
+ Jdbc {
+ driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
+ url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
+ username = "SA"
+ password = "Y.sa123456"
+ query = "select * from full_types_jdbc"
+ partition_column = "id"
+ partition_num = 10
+ }
}
transform {
- # If you would like to get more information about how to configure
seatunnel and see full list of transform plugins,
- # please go to https://seatunnel.apache.org/docs/transforms/sql
+ # 如果你想了解更多关于如何配置 seatunnel 的信息，并查看转换插件的完整列表，
+ # 请前往 https://seatunnel.apache.org/docs/transforms/sql
}
sink {
- Console {}
+ Console {}
}
-
```
-### 分片并行读取简单示例
-
-> 这是一个快速并行读取数据的分片
+### 整库多表读取
-```
+```hocon
env {
- # 您可以在这里设置引擎配置
- parallelism = 10
+ parallelism = 4
+ job.mode = "BATCH"
}
source {
- # 这是一个示例源插件 **仅用于测试和演示源插件功能**
Jdbc {
- driver = com.microsoft.sqlserver.jdbc.SQLServerDriver
+ driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
- username = SA
+ username = "SA"
password = "Y.sa123456"
- query = "select * from column_type_test.dbo.full_types_jdbc"
- # 并行分片读取字段
- partition_column = "id"
- # 分片数量
- partition_num = 10
+ table_list = [
+ { table_path = "column_type_test.dbo.full_types_jdbc" }
+ { table_path = "column_type_test.dbo.orders", query = "select id, name,
status from column_type_test.dbo.orders" }
+ ]
+ where_condition = "where id > 0"
+ split.size = 10000
+ }
+}
+transform {
+ Sql {
+ plugin_input = "Jdbc"
+ plugin_output = "tmp_id_name"
+ query = "select id, name from full_types_jdbc"
}
- # 如果你想了解更多关于如何配置 seatunnel 的信息，并查看源插件的完整列表，
- # 请前往 https://seatunnel.apache.org/docs/connectors/source/Jdbc
}
+sink {
+ Console {}
+}
+```
+
+### 自定义列裁剪示例
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Jdbc {
+ driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
+ url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
+ username = "SA"
+ password = "Y.sa123456"
+ table_path = "column_type_test.dbo.full_types_jdbc"
+ query = "select id, name from column_type_test.dbo.full_types_jdbc"
+ }
+}
transform {
- # 如果你想了解更多关于如何配置 seatunnel 的信息，并查看转换插件的完整列表，
- # 请前往 https://seatunnel.apache.org/docs/transforms/sql
}
sink {
- Console {}
- # 如果你想了解更多关于如何配置 seatunnel 的信息，并查看汇插件的完整列表，
- # 请前往 https://seatunnel.apache.org/docs/connectors/sink/Jdbc
+ Jdbc {
+ driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
+ url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
+ username = "SA"
+ password = "Y.sa123456"
+ generate_sink_sql = true
+ database = "column_type_test"
+ table = "dbo.full_types_jdbc"
+ }
}
```
