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 7922aa20a1 [Docs][Connector-V2] Improve MySQL Oracle and TiDB CDC docs
(#11322)
7922aa20a1 is described below
commit 7922aa20a191a84e346e23e754bcec2b0129ec38
Author: Daniel Carter <[email protected]>
AuthorDate: Sat Jul 4 09:52:27 2026 +0800
[Docs][Connector-V2] Improve MySQL Oracle and TiDB CDC docs (#11322)
Co-authored-by: DanielCarter-stack
<[email protected]>
---
docs/en/connectors/source/MySQL-CDC.md | 64 +++++++++++++++++++++++---
docs/en/connectors/source/Oracle-CDC.md | 81 +++++++++++++++++++++------------
docs/en/connectors/source/TiDB-CDC.md | 49 +++++++++++++++-----
docs/zh/connectors/source/MySQL-CDC.md | 64 +++++++++++++++++++++++---
docs/zh/connectors/source/Oracle-CDC.md | 79 +++++++++++++++++++++-----------
docs/zh/connectors/source/TiDB-CDC.md | 55 +++++++++++++++-------
6 files changed, 294 insertions(+), 98 deletions(-)
diff --git a/docs/en/connectors/source/MySQL-CDC.md
b/docs/en/connectors/source/MySQL-CDC.md
index 5ee7a0c647..56b09aa7d9 100644
--- a/docs/en/connectors/source/MySQL-CDC.md
+++ b/docs/en/connectors/source/MySQL-CDC.md
@@ -186,23 +186,23 @@ When an initial consistent snapshot is made for large
databases, your establishe
| Name | Type | Required | Default |
Description
[...]
|-------------------------------------------|----------|----------|---------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
| url | String | Yes | - |
The URL of the JDBC connection. Refer to a case:
`jdbc:mysql://localhost:3306/test`.
[...]
-| username | String | Yes | - |
Name of the database to use when connecting to the database server.
[...]
+| username | String | Yes | - |
Username used to connect to the MySQL server.
[...]
| password | String | Yes | - |
Password to use when connecting to the database server.
[...]
| database-names | List | No | - |
Database name of the database to monitor.
[...]
| database-pattern | String | No | .* |
The database names RegEx of the database to capture, for example:
`database_prefix.*`.
[...]
-| table-names | List | Yes | - |
Table name of the database to monitor. The table name needs to include the
database name, for example: `database_name.table_name`
[...]
-| table-pattern | String | Yes | - |
The table names RegEx of the database to capture. The table name needs to
include the database name, for example: `database.*\\.table_.*`
[...]
-| table-names-config | List | No | - |
Table config list. for example: [{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]. The snapshotSplitColumn option must
be configured with a unique key. If a non-unique column is provided, the
configuration is ignored and SeaTunnel automatically selects an appropriate
split column internally.
[...]
+| table-names | List | Conditionally
required | - | Table names to monitor. Each value must include the
database name, for example: `database_name.table_name`. Configure either
`table-names` or `table-pattern`.
[...]
+| table-pattern | String | Conditionally
required | - | Regular expression for table names to capture. Each
matched table name includes the database name, for example:
`database.*\\.table_.*`. Configure either `table-names` or `table-pattern`.
[...]
+| table-names-config | List | No | - |
Per-table config list. For example: `[{"table": "db1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]`. Use this when the table has no
primary key, needs a custom primary key, or needs an explicit snapshot split
column. `snapshotSplitColumn` should be a primary key or unique key. If a
non-unique column is provided, SeaTunnel ignores it and automatically selects
an appropriate split column interna [...]
| startup.mode | Enum | No | INITIAL |
Optional startup mode for MySQL CDC consumer, valid enumerations are `initial`,
`earliest`, `latest` , `specific` and `timestamp`. <br/> `initial`: Synchronize
historical data at startup, and then synchronize incremental data.<br/>
`earliest`: Startup from the earliest offset possible.<br/> `latest`: Startup
from the latest offset.<br/> `specific`: Startup from user-supplied specific
offsets.<br/> `timestamp`: [...]
| startup.specific-offset.file | String | No | - |
Start from the specified binlog file name. **Note, This option is required when
the `startup.mode` option used `specific`.**
[...]
| startup.specific-offset.pos | Long | No | - |
Start from the specified binlog file position. **Note, This option is required
when the `startup.mode` option used `specific`.**
[...]
-| startup.timestamp | Long | No | - |
Start from the specified timestamp. **Note, This option is required when the
`startup.mode` option used `timestamp`.**
[...]
+| startup.timestamp | Long | No | - |
Start from the specified timestamp, in milliseconds since Unix epoch. **Note,
This option is required when the `startup.mode` option uses `timestamp`.**
[...]
| stop.mode | Enum | No | NEVER |
Optional stop mode for MySQL CDC consumer, valid enumerations are `never`,
`latest` or `specific`. <br/> `never`: Real-time job don't stop the
source.<br/> `latest`: Stop from the latest offset.<br/> `specific`: Stop from
user-supplied specific offset.
[...]
| stop.specific-offset.file | String | No | - |
Stop from the specified binlog file name. **Note, This option is required when
the `stop.mode` option used `specific`.**
[...]
| stop.specific-offset.pos | Long | No | - |
Stop from the specified binlog file position. **Note, This option is required
when the `stop.mode` option used `specific`.**
[...]
| snapshot.split.size | Integer | No | 8096 |
The split size (number of rows) of table snapshot, captured tables are split
into multiple splits when read the snapshot of table.
[...]
| snapshot.fetch.size | Integer | No | 1024 |
The maximum fetch size for per poll when read table snapshot.
[...]
-| server-id | String | No | - |
A numeric ID or a numeric ID range of this database client, The numeric ID
syntax is like `5400`, the numeric ID range syntax is like '5400-5408'. <br/>
Every ID must be unique across all currently-running database processes in the
MySQL cluster. This connector joins the <br/> MySQL cluster as another server
(with this unique ID) so it can read the binlog. <br/> By default, a random
number is generated between [...]
+| server-id | String | No | - |
Numeric ID or numeric ID range used by this CDC reader, for example `5400` or
`5400-5408`. Each ID must be unique in the MySQL cluster. When the job has
multiple readers or reads multiple tables in parallel, configure an ID range
large enough for the job. If this option is omitted, SeaTunnel generates a
random ID, but an explicit value is recommended for production.
[...]
| server-time-zone | String | No | UTC |
The session time zone in database server. If not set, then
ZoneId.systemDefault() is used to determine the server time zone.
[...]
| connect.timeout.ms | Duration | No | 30000 |
The maximum time that the connector should wait after trying to connect to the
database server before timing out.
[...]
| connect.max-retries | Integer | No | 3 |
The max retry times that the connector should retry to build database server
connection.
[...]
@@ -375,7 +375,7 @@ source {
}
```
-**Data handling when `drop.column` is excluded. For a retained **NOT NULL**
column the `NULL` write is rejected
+**Data handling when `drop.column` is excluded.** For a retained **NOT NULL**
column the `NULL` write is rejected
by the sink, so excluding `drop.column` for a NOT NULL column that the source
has stopped supplying
will fail at the sink.
@@ -411,6 +411,56 @@ sink {
}
```
+### Start From a Specific Binlog Offset
+
+Use `startup.mode = "specific"` when the first record must be read from a
known binlog file and position.
+
+```hocon
+source {
+ MySQL-CDC {
+ server-id = 5654
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ startup.mode = "specific"
+ startup.specific-offset.file = "mysql-bin.000001"
+ startup.specific-offset.pos = 154
+ }
+}
+```
+
+### Route Multiple Source Tables to JDBC
+
+When one MySQL CDC source reads multiple tables, JDBC sink placeholders can
keep the original table name.
+
+```hocon
+source {
+ MySQL-CDC {
+ plugin_output = "customers_mysql_cdc"
+ server-id = 5652-5660
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.orders", "mysql_cdc.products"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ }
+}
+
+sink {
+ jdbc {
+ plugin_input = "customers_mysql_cdc"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc2"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+ database = "mysql_cdc2"
+ table = "${table_name}"
+ primary_keys = ["${primary_key}"]
+ generate_sink_sql = true
+ }
+}
+```
+
## FAQ
### What MySQL permissions are required for CDC?
diff --git a/docs/en/connectors/source/Oracle-CDC.md
b/docs/en/connectors/source/Oracle-CDC.md
index 5393335b44..956fccdd3a 100644
--- a/docs/en/connectors/source/Oracle-CDC.md
+++ b/docs/en/connectors/source/Oracle-CDC.md
@@ -224,20 +224,17 @@ exit;
| Name | Type | Required | Default |
Description
[...]
|-------------------------------------------|----------|-----------|---------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| url | String | Yes | - |
The URL of the JDBC connection. Refer to a case:
`idbc:oracle:thin:datasource01:1523:xe`.
[...]
-| username | String | Yes | - |
Name of the database to use when connecting to the database server.
[...]
+| url | String | Yes | - |
The URL of the JDBC connection. For example:
`jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB`.
[...]
+| username | String | Yes | - |
Username used to connect to the Oracle database.
[...]
| password | String | Yes | - |
Password to use when connecting to the database server.
[...]
| database-names | List | No | - |
Database name of the database to monitor.
[...]
| schema-names | List | No | - |
Schema name of the database to monitor.
[...]
-| table-names | List | Yes | - |
Table name of the database to monitor. The table name needs to include the
database name, for example: `database_name.table_name`
[...]
-| table-names-config | List | No | - |
Table config list. for example: [{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]
[...]
-| startup.mode | Enum | No | INITIAL |
Optional startup mode for Oracle CDC consumer, valid enumerations are
`initial`, `earliest`, `latest`, `timestamp` and `specific`. <br/> `initial`:
Synchronize historical data at startup, and then synchronize incremental
data.<br/> `earliest`: Startup from the earliest offset possible.<br/>
`latest`: Startup from the latest offset.<br/> `specific`: Startup from
user-supplied specific offsets. [...]
+| table-names | List | Conditionally
required | - | Table names to monitor. Each value should use
`database.schema.table`, for example: `ORCLCDB.DEBEZIUM.FULL_TYPES`. Configure
either `table-names` or `table-pattern`.
[...]
+| table-pattern | String | Conditionally
required | - | Regular expression for table names to capture. Configure
either `table-names` or `table-pattern`.
[...]
+| table-names-config | List | No | - |
Per-table config list. For example: `[{"table":
"ORCLCDB.DEBEZIUM.FULL_TYPES","primaryKeys": ["ID"],"snapshotSplitColumn":
"ID"}]`. Use this when the table has no primary key, needs a custom primary
key, or needs an explicit snapshot split column.
[...]
+| startup.mode | Enum | No | INITIAL |
Optional startup mode for Oracle CDC consumer, valid enumerations are
`initial`, `latest` and `timestamp`. <br/> `initial`: Synchronize historical
data at startup, and then synchronize incremental data.<br/> `latest`: Start
from the latest offset and skip the initial snapshot.<br/> `timestamp`: Start
from the SCN resolved from `startup.timestamp`.
[...]
| startup.timestamp | Long | No | - |
Start from the specified timestamp (milliseconds since Unix epoch). This
timestamp is converted with `server-time-zone` when `startup.mode = timestamp`.
**Note, This option is required when the `startup.mode` option used
`timestamp`.**
[...]
-| startup.specific-offset.file | String | No | - |
Start from the specified binlog file name. **Note, This option is required when
the `startup.mode` option used `specific`.**
[...]
-| startup.specific-offset.pos | Long | No | - |
Start from the specified binlog file position. **Note, This option is required
when the `startup.mode` option used `specific`.**
[...]
-| stop.mode | Enum | No | NEVER |
Optional stop mode for Oracle CDC consumer, valid enumerations are `never`,
`latest` or `specific`. <br/> `never`: Real-time job don't stop the
source.<br/> `latest`: Stop from the latest offset.<br/> `specific`: Stop from
user-supplied specific offset.
[...]
-| stop.specific-offset.file | String | No | - |
Stop from the specified binlog file name. **Note, This option is required when
the `stop.mode` option used `specific`.**
[...]
-| stop.specific-offset.pos | Long | No | - |
Stop from the specified binlog file position. **Note, This option is required
when the `stop.mode` option used `specific`.**
[...]
+| stop.mode | Enum | No | NEVER |
Optional stop mode for Oracle CDC consumer. The only valid value is `never`, so
a streaming Oracle CDC source keeps running until the job is stopped.
[...]
| snapshot.split.size | Integer | No | 8096 |
The split size (number of rows) of table snapshot, captured tables are split
into multiple splits when read the snapshot of table.
[...]
| snapshot.fetch.size | Integer | No | 1024 |
The maximum fetch size for per poll when read table snapshot.
[...]
| server-time-zone | String | No | UTC |
The session time zone in database server. If not set, then
ZoneId.systemDefault() is used to determine the server time zone. This value is
also used when converting `startup.timestamp` to SCN. Set it explicitly when
database time zone and JVM time zone are different.
[...]
@@ -295,12 +292,16 @@ source {
Oracle-CDC {
plugin_output = "customers"
username = "system"
- password = "oracle"
- database-names = ["XE"]
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
schema-names = ["DEBEZIUM"]
- table-names = ["XE.DEBEZIUM.FULL_TYPES", "XE.DEBEZIUM.FULL_TYPES2"]
- url = "jdbc:oracle:thin:@oracle-host:1521:xe"
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES",
"ORCLCDB.DEBEZIUM.FULL_TYPES2"]
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
source.reader.close.timeout = 120000
+ connection.pool.size = 1
+ debezium {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
}
}
```
@@ -313,11 +314,11 @@ source {
plugin_output = "customers"
use_select_count = true
username = "system"
- password = "oracle"
- database-names = ["XE"]
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
schema-names = ["DEBEZIUM"]
- table-names = ["XE.DEBEZIUM.FULL_TYPES"]
- url = "jdbc:oracle:thin:system/oracle@oracle-host:1521:xe"
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES"]
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
source.reader.close.timeout = 120000
}
}
@@ -332,11 +333,11 @@ source {
plugin_output = "customers"
skip_analyze = true
username = "system"
- password = "oracle"
- database-names = ["XE"]
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
schema-names = ["DEBEZIUM"]
- table-names = ["XE.DEBEZIUM.FULL_TYPES"]
- url = "jdbc:oracle:thin:system/oracle@oracle-host:1521:xe"
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES"]
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
source.reader.close.timeout = 120000
}
}
@@ -348,16 +349,16 @@ source {
source {
Oracle-CDC {
plugin_output = "customers"
- url = "jdbc:oracle:thin:system/oracle@oracle-host:1521:xe"
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
source.reader.close.timeout = 120000
username = "system"
- password = "oracle"
- database-names = ["XE"]
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
schema-names = ["DEBEZIUM"]
- table-names = ["XE.DEBEZIUM.FULL_TYPES"]
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES"]
table-names-config = [
{
- table = "XE.DEBEZIUM.FULL_TYPES"
+ table = "ORCLCDB.DEBEZIUM.FULL_TYPES"
primaryKeys = ["ID"]
}
]
@@ -365,6 +366,30 @@ source {
}
```
+### Start From Timestamp
+
+Use `startup.mode = "timestamp"` to start from the Oracle SCN resolved from a
Unix timestamp in milliseconds.
+
+```hocon
+source {
+ Oracle-CDC {
+ plugin_output = "customers"
+ username = "system"
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
+ schema-names = ["DEBEZIUM"]
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES"]
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
+ startup.mode = "timestamp"
+ startup.timestamp = 1700000000000
+ server-time-zone = "UTC"
+ debezium {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
+ }
+}
+```
+
### Schema change event filtering
When `schema-changes.enabled = true`, you can further control which schema
change event types are
@@ -397,7 +422,7 @@ source {
}
```
-**Data handling when `drop.column` is excluded. For a retained **NOT NULL**
column the `NULL` write is rejected
+**Data handling when `drop.column` is excluded.** For a retained **NOT NULL**
column the `NULL` write is rejected
by the sink, so excluding `drop.column` for a NOT NULL column that the source
has stopped supplying
will fail at the sink.
diff --git a/docs/en/connectors/source/TiDB-CDC.md
b/docs/en/connectors/source/TiDB-CDC.md
index 11f682d997..aac65a24ce 100644
--- a/docs/en/connectors/source/TiDB-CDC.md
+++ b/docs/en/connectors/source/TiDB-CDC.md
@@ -42,7 +42,7 @@ describes how to set up the TiDB CDC connector to snapshot
data and capture stre
> 1. You need to ensure that the [jdbc driver jar
> package](https://mvnrepository.com/artifact/mysql/mysql-connector-java) and
> the [tikv-client-java jar
> package](https://mvnrepository.com/artifact/org.tikv/tikv-client-java/3.2.0)
> has been placed in directory `${SEATUNNEL_HOME}/lib/`.
-Please download and put Mysql driver and tikv-java-client in
`${SEATUNNEL_HOME}/lib/` dir. For example: cp mysql-connector-java-xxx.jar
`$SEATUNNEL_HOME/lib/`
+Please download and put the MySQL driver and tikv-java-client in the directory
required by your engine.
## Data Type Mapping
@@ -68,12 +68,12 @@ Please download and put Mysql driver and tikv-java-client
in `${SEATUNNEL_HOME}/
| Name | Type | Required | Default | Description
|
|-------------------------|---------|----------|---------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| url | String | Yes | - | The URL of the JDBC
connection. Refer to a case: `jdbc:mysql://tidb0:4000/inventory`.
|
-| username | String | Yes | - | Name of the
database to use when connecting to the database server.
|
+| username | String | Yes | - | Username used to
connect to the TiDB server.
|
| password | String | Yes | - | Password to use
when connecting to the database server.
|
| pd-addresses | String | Yes | - | TiKV cluster's PD
address
|
| database-name | String | Yes | - | Database name of
the database to monitor.
|
-| table-name | String | Yes | - | Table name of the
database to monitor. The table name needs to include the database name.
|
-| startup.mode | Enum | No | INITIAL | Optional startup
mode for TiDB CDC consumer, valid enumerations are `initial`, `earliest`,
`latest` and `specific`. <br/> `initial`: Synchronize historical data at
startup, and then synchronize incremental data.<br/> `earliest`: Startup from
the earliest offset possible.<br/> `latest`: Startup from the latest
offset.<br/> `specific`: Startup from user-supplied specific offsets. |
+| table-name | String | Yes | - | Table name to
monitor in `database-name`. Do not include the database name here.
|
+| startup.mode | Enum | No | INITIAL | Optional startup
mode for TiDB CDC consumer, valid enumerations are `initial`, `earliest` and
`latest`. <br/> `initial`: Synchronize historical data at startup, and then
synchronize incremental data.<br/> `earliest`: Startup from the earliest
available offset.<br/> `latest`: Startup from the latest offset and skip the
initial snapshot.
[...]
| batch-size-per-scan | Int | No | 1000 | Size per scan.
|
| tikv.grpc.timeout_in_ms | Long | No | - | TiKV GRPC timeout
in ms.
|
| tikv.grpc.scan_timeout_in_ms | Long | No | - | TiKV GRPC scan
timeout in ms.
|
@@ -84,7 +84,7 @@ Please download and put Mysql driver and tikv-java-client in
`${SEATUNNEL_HOME}/
### Simple
-```
+```hocon
env {
parallelism = 1
job.mode = "STREAMING"
@@ -95,14 +95,14 @@ source {
# This is a example source plugin **only for test and demonstrate the
feature source plugin**
TiDB-CDC {
plugin_output = "products_tidb_cdc"
- url = "jdbc:mysql://tidb0:4000/inventory"
+ url = "jdbc:mysql://tidb0:4000/tidb_cdc"
driver = "com.mysql.cj.jdbc.Driver"
tikv.grpc.timeout_in_ms = 20000
pd-addresses = "pd0:2379"
username = "root"
password = ""
- database-name = "inventory"
- table-name = "products"
+ database-name = "tidb_cdc"
+ table-name = "tidb_cdc_e2e_source_table"
}
}
@@ -112,18 +112,43 @@ transform {
sink {
jdbc {
plugin_input = "products_tidb_cdc"
- url = "jdbc:mysql://tidb0:4000/inventory"
+ url = "jdbc:mysql://tidb0:4000/tidb_cdc"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = ""
- database = "inventory"
- table = "products_sink"
+ database = "tidb_cdc"
+ table = "tidb_cdc_e2e_sink_table"
generate_sink_sql = true
primary_keys = ["id"]
}
}
```
+### Start From Latest Offset
+
+Use `startup.mode = "latest"` when only new changes are needed and the
historical snapshot should be skipped.
+
+```hocon
+source {
+ TiDB-CDC {
+ url = "jdbc:mysql://tidb0:4000/tidb_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ pd-addresses = "pd0:2379"
+ username = "root"
+ password = ""
+ database-name = "tidb_cdc"
+ table-name = "tidb_cdc_e2e_source_table"
+ startup.mode = "latest"
+ }
+}
+```
+
+## Notes
+
+- TiDB CDC reads one table per source block. Use multiple `TiDB-CDC` source
blocks if one job needs to capture multiple tables.
+- `startup.mode = "specific"` is not a valid TiDB CDC option. Use `initial`,
`earliest`, or `latest`.
+- Tune `tikv.grpc.*` and `tikv.batch_*_concurrency` only when the default TiKV
client settings are not enough for your cluster.
+
## Changelog
<ChangeLog />
diff --git a/docs/zh/connectors/source/MySQL-CDC.md
b/docs/zh/connectors/source/MySQL-CDC.md
index b8c27ecc0f..ee4f30d27d 100644
--- a/docs/zh/connectors/source/MySQL-CDC.md
+++ b/docs/zh/connectors/source/MySQL-CDC.md
@@ -185,23 +185,23 @@ show variables where variable_name in ('log_bin',
'binlog_format', 'binlog_row_i
| 参数名称 | 类型 | 是否必须 | 默认值 | 描述
|
|-------------------------------------------|----------|------|---------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| url | String | 是 | - |
JDBC连接的URL. 例如: `jdbc:mysql://localhost:3306/test`.
|
-| username | String | 是 | - |
用来连接到数据库服务的数据库名称.
|
+| username | String | 是 | - | 连接
MySQL 服务时使用的用户名.
|
| password | String | 是 | - |
连接到数据库服务所使用的密码.
|
| database-names | List | 否 | - |
要监控的数据库名称.
|
| database-pattern | String | 否 | .* |
要捕获的数据库名称的正则表达式, 例如: `database_prefix.*`.
|
-| table-names | List | 是 | - |
要监控的表名. 表名需要包括库名, 例如: `database_name.table_name`
|
-| table-pattern | String | 是 | - |
要捕获的表名称的正则表达式. 表名需要包括库名, 例如: `database.*\\.table_.*`
|
-| table-names-config | List | 否 | - |
表配置的列表集合. 例如: [{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]. snapshotSplitColumn
选项必须配置为唯一键(主键或唯一索引). 如果指定了非唯一列，该配置将被忽略，SeaTunnel 会在内部自动选择合适的拆分列.
|
+| table-names | List | 条件必填 | - |
要监控的表名，每个表名都需要包含库名，例如：`database_name.table_name`。`table-names` 和
`table-pattern` 二选一配置。
|
+| table-pattern | String | 条件必填 | - |
要捕获的表名正则表达式，匹配到的表名需要包含库名，例如：`database.*\\.table_.*`。`table-names` 和
`table-pattern` 二选一配置。
|
+| table-names-config | List | 否 | - |
按表单独配置。例如：`[{"table": "db1.table1","primaryKeys":
["key1"],"snapshotSplitColumn":
"key2"}]`。当表没有主键、需要自定义主键，或需要指定快照拆分列时使用。`snapshotSplitColumn`
应该是主键或唯一键；如果指定了非唯一列，SeaTunnel 会忽略该配置并自动选择合适的拆分列。
|
| startup.mode | Enum | 否 | INITIAL |
MySQL CDC 消费者的可选启动模式, 有效枚举值为 `initial`, `earliest`, `latest` , `specific` 和
`timestamp`. <br/> `initial`: 启动时同步历史数据, 然后同步增量数据.<br/> `earliest`:
从尽可能最早的偏移量开始启动.<br/> `latest`: 从最近的偏移量启动.<br/> `specific`:
从用户提供的特定偏移量开始启动.<br/> `timestamp`: 从用户提供的特定时间戳开始启动. |
| startup.specific-offset.file | String | 否 | - |
从指定的binlog日志文件名开始. **注意, 当使用 `startup.mode` 选项为 `specific` 时，此选项为必填项.**
|
| startup.specific-offset.pos | Long | 否 | - |
从指定的binlog日志文件位置开始. **注意, 当使用 `startup.mode` 选项为 `specific` 时，此选项为必填项.**
|
-| startup.timestamp | Long | No | - |
从指定的binlog时间戳文件位置开始. **注意, 当使用 `startup.mode` 选项为 `timestamp` 时，此选项为必填项.**
|
+| startup.timestamp | Long | 否 | - |
从指定时间戳启动，单位为 Unix 纪元以来的毫秒数。**注意，当 `startup.mode` 为 `timestamp` 时，此选项必填。**
|
| stop.mode | Enum | 否 | NEVER |
MySQL CDC 消费者的可选停止模式, 有效枚举值为 `never`, `latest` 和 `specific`. <br/> `never`:
实时任务一直运行不停止.<br/> `latest`: 从最新的偏移量处停止.<br/> `specific`: 从用户提供的特定偏移量处停止.
|
| stop.specific-offset.file | String | 否 | - |
从指定的binlog日志文件名停止. **注意, 当使用 `stop.mode` 选项为 `specific` 时，此选项为必填项.**
|
| stop.specific-offset.pos | Long | 否 | - |
从指定的binlog日志文件位置停止. **注意, 当使用 `stop.mode` 选项为 `specific` 时，此选项为必填项.**
|
| snapshot.split.size | Integer | 否 | 8096 |
表快照的分割大小（行数）,读取表的快照时,被捕获的表会被分割成多个分割块.
|
| snapshot.fetch.size | Integer | 否 | 1024 |
每次轮询读取表快照时的最大获取大小.
|
-| server-id | String | 否 | - |
此数据库客户端的数字 ID 或数字 ID 范围, 数字 ID 的语法如 `5400`, 数字 ID 范围的语法如 '5400-5408'. <br/> 每个
ID 在 MySQL 集群中所有当前正在运行的数据库进程里必须是唯一的. 此连接加入 <br/> MySQL服务以另外一个服务的身份 (带有此唯一 ID)
以便于能够读取binlog. <br/> 默认情况下, 会生成一个介于 6500 到 2,148,492,146 之间的数字, 然而我们建议设置一个明确的值.
|
+| server-id | String | 否 | - | 此
CDC 读取器使用的数字 ID 或数字 ID 范围，例如 `5400` 或 `5400-5408`。每个 ID 在 MySQL
集群中必须唯一。当任务有多个读取并发或并行读取多张表时，请配置足够大的 ID 范围。未配置时 SeaTunnel 会随机生成 ID，但生产环境建议显式配置。 |
| server-time-zone | String | 否 | UTC |
数据库服务中的会话时区. 如果没设置, 使用 ZoneId.systemDefault() 来确定服务的时区.
|
| connect.timeout.ms | Duration | 否 | 30000 |
连接器在尝试连接数据库服务器后，在超时之前应等待的最长时间.
|
| connect.max-retries | Integer | 否 | 3 |
连接器在构建数据库服务器连接时应重试的最大重试次数.
|
@@ -374,7 +374,7 @@ source {
}
```
-**排除 `drop.column` 时的数据处理方式。对于被保留的 **NOT NULL** 列，写入 `NULL` 会被 sink
拒绝，因此对一个源端已不再供数的
+**排除 `drop.column` 时的数据处理方式。** 对于被保留的 **NOT NULL** 列，写入 `NULL` 会被 sink
拒绝，因此对一个源端已不再供数的
NOT NULL 列排除 `drop.column` 会在 sink 端失败。
### 表名支持正则以读取多个表
@@ -408,6 +408,56 @@ sink {
}
```
+### 从指定 Binlog 位置启动
+
+当需要从明确的 binlog 文件和位置开始读取时，可以使用 `startup.mode = "specific"`。
+
+```hocon
+source {
+ MySQL-CDC {
+ server-id = 5654
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ startup.mode = "specific"
+ startup.specific-offset.file = "mysql-bin.000001"
+ startup.specific-offset.pos = 154
+ }
+}
+```
+
+### 多表读取后写入 JDBC
+
+当一个 MySQL CDC source 读取多张表时，JDBC sink 可以使用占位符保留原始表名。
+
+```hocon
+source {
+ MySQL-CDC {
+ plugin_output = "customers_mysql_cdc"
+ server-id = 5652-5660
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.orders", "mysql_cdc.products"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ }
+}
+
+sink {
+ jdbc {
+ plugin_input = "customers_mysql_cdc"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc2"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+ database = "mysql_cdc2"
+ table = "${table_name}"
+ primary_keys = ["${primary_key}"]
+ generate_sink_sql = true
+ }
+}
+```
+
## 常见问题
### MySQL CDC 需要哪些权限？
diff --git a/docs/zh/connectors/source/Oracle-CDC.md
b/docs/zh/connectors/source/Oracle-CDC.md
index 534144ce30..e87827059e 100644
--- a/docs/zh/connectors/source/Oracle-CDC.md
+++ b/docs/zh/connectors/source/Oracle-CDC.md
@@ -223,20 +223,17 @@ exit;
| 参数名称 | 类型 | 是否必选 | 默认值 | 描述
[...]
|-------------------------------------------|----------|--------|---------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| url | String | 是 | - |
JDBC 连接的 URL。例如：`jdbc:oracle:thin:datasource01:1523:xe`。
[...]
+| url | String | 是 | - |
JDBC 连接的 URL，例如：`jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB`。
[...]
| username | String | 是 | - |
连接数据库服务器时使用的数据库用户名。
[...]
| password | String | 是 | - |
连接数据库服务器时使用的数据库密码。
[...]
| database-names | List | 否 | - |
要监控的数据库名称。
[...]
| schema-names | List | 否 | - |
要监控的数据库 Schema 名称。
[...]
-| table-names | List | 是 | - |
要监控的数据库表名。表名需要包含数据库名，例如：`database_name.table_name`
[...]
-| table-names-config | List | 否 | - |
表配置列表。例如：`[{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]`
[...]
-| startup.mode | Enum | 否 | INITIAL |
Oracle CDC 使用者的可选启动模式，有效枚举值为 `initial`、`earliest`、`latest`、`timestamp` 和
`specific`。<br/> `initial`：启动时同步历史数据，然后同步增量数据。<br/>
`earliest`：从尽可能早的偏移量启动。<br/> `latest`：从最新的偏移量启动。<br/> `specific`：从用户提供的特定偏移量启动。
|
+| table-names | List | 条件必填 | - |
要监控的数据库表名，建议使用 `database.schema.table`
格式，例如：`ORCLCDB.DEBEZIUM.FULL_TYPES`。`table-names` 和 `table-pattern` 二选一配置。
[...]
+| table-pattern | String | 条件必填 | - |
要捕获的表名正则表达式。`table-names` 和 `table-pattern` 二选一配置。
[...]
+| table-names-config | List | 否 | - |
按表单独配置。例如：`[{"table": "ORCLCDB.DEBEZIUM.FULL_TYPES","primaryKeys":
["ID"],"snapshotSplitColumn": "ID"}]`。当表没有主键、需要自定义主键，或需要指定快照拆分列时使用。
[...]
+| startup.mode | Enum | 否 | INITIAL |
Oracle CDC 使用者的可选启动模式，有效枚举值为 `initial`、`latest` 和 `timestamp`。<br/>
`initial`：启动时同步历史数据，然后同步增量数据。<br/> `latest`：从最新偏移量启动，并跳过初始快照。<br/>
`timestamp`：从 `startup.timestamp` 解析出的 SCN 启动。
|
| startup.timestamp | Long | 否 | - |
从指定的时间戳（自 Unix 纪元以来的毫秒数）启动。当 `startup.mode = timestamp` 时，该时间戳会按
`server-time-zone` 转换。**注意，当 `startup.mode` 选项使用 `timestamp` 时，此选项是必需的。**
[...]
-| startup.specific-offset.file | String | 否 | - |
从指定的 binlog 文件名启动。**注意，当 `startup.mode` 选项使用 `specific` 时，此选项是必需的。**
[...]
-| startup.specific-offset.pos | Long | 否 | - |
从指定的 binlog 文件位置启动。**注意，当 `startup.mode` 选项使用 `specific` 时，此选项是必需的。**
[...]
-| stop.mode | Enum | 否 | NEVER |
Oracle CDC 使用者的可选停止模式，有效枚举值为 `never`、`latest` 或 `specific`。<br/>
`never`：实时任务不停止源。<br/> `latest`：从最新的偏移量停止。<br/> `specific`：从用户提供的特定偏移量停止。
[...]
-| stop.specific-offset.file | String | 否 | - |
从指定的 binlog 文件名停止。**注意，当 `stop.mode` 选项使用 `specific` 时，此选项是必需的。**
[...]
-| stop.specific-offset.pos | Long | 否 | - |
从指定的 binlog 文件位置停止。**注意，当 `stop.mode` 选项使用 `specific` 时，此选项是必需的。**
[...]
+| stop.mode | Enum | 否 | NEVER |
Oracle CDC 使用者的可选停止模式。当前唯一有效值是 `never`，因此流式 Oracle CDC source 会一直运行，直到任务被停止。
[...]
| snapshot.split.size | Integer | 否 | 8096 |
表快照的拆分大小（行数），在读取表快照时，捕获的表将被拆分为多个拆分块。
[...]
| snapshot.fetch.size | Integer | 否 | 1024 |
读取表快照时每次轮询的最大获取大小。
[...]
| server-time-zone | String | 否 | UTC |
数据库服务器中的会话时区。如果未设置，则使用 ZoneId.systemDefault() 来确定服务器时区。该参数也用于将
`startup.timestamp` 转换为 SCN。若数据库时区与 JVM 时区不同，建议显式配置。
[...]
@@ -294,12 +291,16 @@ source {
Oracle-CDC {
plugin_output = "customers"
username = "system"
- password = "oracle"
- database-names = ["XE"]
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
schema-names = ["DEBEZIUM"]
- table-names = ["XE.DEBEZIUM.FULL_TYPES", "XE.DEBEZIUM.FULL_TYPES2"]
- url = "jdbc:oracle:thin:@oracle-host:1521:xe"
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES",
"ORCLCDB.DEBEZIUM.FULL_TYPES2"]
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
source.reader.close.timeout = 120000
+ connection.pool.size = 1
+ debezium {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
}
}
```
@@ -312,11 +313,11 @@ source {
plugin_output = "customers"
use_select_count = true
username = "system"
- password = "oracle"
- database-names = ["XE"]
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
schema-names = ["DEBEZIUM"]
- table-names = ["XE.DEBEZIUM.FULL_TYPES"]
- url = "jdbc:oracle:thin:system/oracle@oracle-host:1521:xe"
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES"]
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
source.reader.close.timeout = 120000
}
}
@@ -331,11 +332,11 @@ source {
plugin_output = "customers"
skip_analyze = true
username = "system"
- password = "oracle"
- database-names = ["XE"]
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
schema-names = ["DEBEZIUM"]
- table-names = ["XE.DEBEZIUM.FULL_TYPES"]
- url = "jdbc:oracle:thin:system/oracle@oracle-host:1521:xe"
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES"]
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
source.reader.close.timeout = 120000
}
}
@@ -347,16 +348,16 @@ source {
source {
Oracle-CDC {
plugin_output = "customers"
- url = "jdbc:oracle:thin:system/oracle@oracle-host:1521:xe"
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
source.reader.close.timeout = 120000
username = "system"
- password = "oracle"
- database-names = ["XE"]
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
schema-names = ["DEBEZIUM"]
- table-names = ["XE.DEBEZIUM.FULL_TYPES"]
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES"]
table-names-config = [
{
- table = "XE.DEBEZIUM.FULL_TYPES"
+ table = "ORCLCDB.DEBEZIUM.FULL_TYPES"
primaryKeys = ["ID"]
}
]
@@ -364,6 +365,30 @@ source {
}
```
+### 从时间戳启动
+
+使用 `startup.mode = "timestamp"` 时，Oracle CDC 会根据毫秒级 Unix 时间戳解析对应的 Oracle SCN
并从该位置启动。
+
+```hocon
+source {
+ Oracle-CDC {
+ plugin_output = "customers"
+ username = "system"
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
+ schema-names = ["DEBEZIUM"]
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES"]
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
+ startup.mode = "timestamp"
+ startup.timestamp = 1700000000000
+ server-time-zone = "UTC"
+ debezium {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
+ }
+}
+```
+
### Schema change 事件过滤
当 `schema-changes.enabled = true` 时，可通过 `schema-changes.include` /
`schema-changes.exclude` 进一步
@@ -396,7 +421,7 @@ source {
}
```
-**排除 `drop.column` 时的数据处理方式。对于被保留的 **NOT NULL** 列，写入 `NULL` 会被 sink
拒绝，因此对一个源端已不再供数的
+**排除 `drop.column` 时的数据处理方式。** 对于被保留的 **NOT NULL** 列，写入 `NULL` 会被 sink
拒绝，因此对一个源端已不再供数的
NOT NULL 列排除 `drop.column` 会在 sink 端失败。
### 支持以兼容 debezium 的格式发送到 kafka
diff --git a/docs/zh/connectors/source/TiDB-CDC.md
b/docs/zh/connectors/source/TiDB-CDC.md
index e3f358e33f..dde40d2922 100644
--- a/docs/zh/connectors/source/TiDB-CDC.md
+++ b/docs/zh/connectors/source/TiDB-CDC.md
@@ -29,7 +29,7 @@ TiDB-CDC连接器允许从 TiDB 数据库读取快照数据和增量数据。本
| MySQL | <li> [MySQL](https://dev.mysql.com/doc): 5.5, 5.6, 5.7,
8.0.x </li><li> [RDS MySQL](https://www.aliyun.com/product/rds/mysql): 5.6,
5.7, 8.0.x </li> | com.mysql.cj.jdbc.Driver |
https://mvnrepository.com/artifact/mysql/mysql-connector-java/8.0.28 |
| tikv-client-java | 3.2.0
| - |
https://mvnrepository.com/artifact/org.tikv/tikv-client-java/3.2.0 |
-## Using Dependency
+## 依赖
### 安装驱动
@@ -41,11 +41,7 @@ TiDB-CDC连接器允许从 TiDB 数据库读取快照数据和增量数据。本
> 1. 你需要确保 [jdbc 驱动 jar
> 包](https://mvnrepository.com/artifact/mysql/mysql-connector-java) 和
> [tikv-client-java jar
> 包](https://mvnrepository.com/artifact/org.tikv/tikv-client-java/3.2.0)
> 已经放在目录 `${SEATUNNEL_HOME}/lib/` 。
-请下载 Mysql 驱动和 tikv-java-client 并将其放在 `${SEATUNNEL_HOME}/lib/` 目录中。例如：
-
-```bash
-cp mysql-connector-java-xxx.jar ${SEATUNNEL_HOME}/lib/
-```
+请下载 MySQL 驱动和 tikv-java-client，并按所使用引擎的要求放到对应目录。
## 数据类型映射
@@ -75,8 +71,8 @@ cp mysql-connector-java-xxx.jar ${SEATUNNEL_HOME}/lib/
| password | String | 是 | - | 连接数据库服务器时使用的密码。
|
| pd-addresses | String | 是 | - | TiKV 集群的 PD 地址。
|
| database-name | String | 是 | - | 要监控的数据库名称。
|
-| table-name | String | 是 | - | 要监控的表名称。表名称需要包含数据库名称。
|
-| startup.mode | Enum | 否 | INITIAL | TiDB CDC 消费器的可选启动模式，可选值有
`initial`、`earliest`、`latest` 和
`specific`。<br/>`initial`：启动时同步历史数据，然后同步增量数据。<br/>`earliest`：从最早的可用偏移量开始启动。<br/>`latest`：从最新的偏移量开始启动。<br/>`specific`：从用户提供的特定偏移量开始启动。
|
+| table-name | String | 是 | - | 要监控的表名，属于 `database-name`
指定的数据库。这里不要再包含数据库名。
|
+| startup.mode | Enum | 否 | INITIAL | TiDB CDC 消费器的可选启动模式，可选值有
`initial`、`earliest` 和
`latest`。<br/>`initial`：启动时同步历史数据，然后同步增量数据。<br/>`earliest`：从最早的可用偏移量开始启动。<br/>`latest`：从最新的偏移量开始启动，并跳过初始快照。
|
| batch-size-per-scan | Int | 否 | 1000 | 每次扫描的大小。
|
| tikv.grpc.timeout_in_ms | Long | 否 | - | TiKV GRPC 超时时间（毫秒）。
|
| tikv.grpc.scan_timeout_in_ms | Long | 否 | - | TiKV GRPC
扫描超时时间（毫秒）。
|
@@ -87,7 +83,7 @@ cp mysql-connector-java-xxx.jar ${SEATUNNEL_HOME}/lib/
### 简单示例
-```
+```hocon
env {
parallelism = 1
job.mode = "STREAMING"
@@ -97,14 +93,14 @@ env {
source {
TiDB-CDC {
plugin_output = "products_tidb_cdc"
- url = "jdbc:mysql://tidb0:4000/inventory"
+ url = "jdbc:mysql://tidb0:4000/tidb_cdc"
driver = "com.mysql.cj.jdbc.Driver"
tikv.grpc.timeout_in_ms = 20000
pd-addresses = "pd0:2379"
username = "root"
password = ""
- database-name = "inventory"
- table-name = "products"
+ database-name = "tidb_cdc"
+ table-name = "tidb_cdc_e2e_source_table"
}
}
@@ -114,18 +110,43 @@ transform {
sink {
jdbc {
plugin_input = "products_tidb_cdc"
- url = "jdbc:mysql://tidb0:4000/inventory"
+ url = "jdbc:mysql://tidb0:4000/tidb_cdc"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = ""
- database = "inventory"
- table = "products_sink"
+ database = "tidb_cdc"
+ table = "tidb_cdc_e2e_sink_table"
generate_sink_sql = true
primary_keys = ["id"]
}
}
```
+### 从最新位置启动
+
+如果只需要读取新的变更、不需要同步历史快照，可以使用 `startup.mode = "latest"`。
+
+```hocon
+source {
+ TiDB-CDC {
+ url = "jdbc:mysql://tidb0:4000/tidb_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ pd-addresses = "pd0:2379"
+ username = "root"
+ password = ""
+ database-name = "tidb_cdc"
+ table-name = "tidb_cdc_e2e_source_table"
+ startup.mode = "latest"
+ }
+}
+```
+
+## 注意事项
+
+- TiDB CDC 每个 source 块读取一张表；如果一个任务要读取多张表，请配置多个 `TiDB-CDC` source 块。
+- `startup.mode = "specific"` 不是 TiDB CDC 的有效配置值，请使用 `initial`、`earliest` 或
`latest`。
+- 只有默认 TiKV 客户端设置不能满足集群需求时，才需要调整 `tikv.grpc.*` 和 `tikv.batch_*_concurrency` 参数。
+
## 变更日志
-<ChangeLog />
\ No newline at end of file
+<ChangeLog />
