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 47084f2534 [Docs][Connector-V2] Improve Neo4j Oracle CDC and Druid
docs (#11342)
47084f2534 is described below
commit 47084f253498b26903062d1fe513592fb89065ac
Author: Daniel Carter <[email protected]>
AuthorDate: Mon Jul 6 22:32:13 2026 +0800
[Docs][Connector-V2] Improve Neo4j Oracle CDC and Druid docs (#11342)
Co-authored-by: DanielCarter-stack
<[email protected]>
---
docs/en/connectors/sink/Druid.md | 5 ++++
docs/en/connectors/sink/Neo4j.md | 2 ++
docs/en/connectors/source/Neo4j.md | 3 ++
docs/en/connectors/source/Oracle-CDC.md | 53 +++++++++++++++++++++++++++++++++
docs/zh/connectors/sink/Druid.md | 5 ++++
docs/zh/connectors/sink/Neo4j.md | 2 ++
docs/zh/connectors/source/Neo4j.md | 3 ++
docs/zh/connectors/source/Oracle-CDC.md | 53 +++++++++++++++++++++++++++++++++
8 files changed, 126 insertions(+)
diff --git a/docs/en/connectors/sink/Druid.md b/docs/en/connectors/sink/Druid.md
index 2fbbca640d..229dff70ad 100644
--- a/docs/en/connectors/sink/Druid.md
+++ b/docs/en/connectors/sink/Druid.md
@@ -75,6 +75,10 @@ Rows are flushed when the buffered row count reaches
`batchSize`. Remaining rows
The sink is designed for append-style batch ingestion. It does not interpret
CDC update/delete row kinds as Druid upserts or deletes.
+Only the data types listed in [Data Type Mapping](#data-type-mapping) are
supported. Other SeaTunnel types fail during write planning.
+
+Because the connector submits inline CSV data to Druid, string values should
not contain raw commas or line breaks unless they are normalized before the
Druid sink.
+
## Example
### Write One Table
@@ -119,6 +123,7 @@ sink {
Druid {
coordinatorUrl = "router:8888"
datasource = "testDataSource"
+ batchSize = 10000
}
}
```
diff --git a/docs/en/connectors/sink/Neo4j.md b/docs/en/connectors/sink/Neo4j.md
index bfe288a1a0..ca96220730 100644
--- a/docs/en/connectors/sink/Neo4j.md
+++ b/docs/en/connectors/sink/Neo4j.md
@@ -39,6 +39,8 @@ It supports one-row-at-a-time writes and batch writes with
Cypher `UNWIND`.
- In `ONE_BY_ONE` mode, `queryParamPosition` maps each Cypher placeholder to a
field position in the input row.
- In `BATCH` mode, the query should use `UNWIND $batch AS row`. The connector
passes the rows through the `batch` variable.
- `queryParamPosition` is still required by the connector configuration in
`BATCH` mode, even though the batch query reads values from `row`.
+- Field positions in `queryParamPosition` start from `0`, following the input
schema field order.
+- In `BATCH` mode, each `row` entry uses the input field names, so the names
referenced in the Cypher statement must match the upstream schema.
## Write One Row At A Time
diff --git a/docs/en/connectors/source/Neo4j.md
b/docs/en/connectors/source/Neo4j.md
index 0ce4c4906e..3341df1522 100644
--- a/docs/en/connectors/source/Neo4j.md
+++ b/docs/en/connectors/source/Neo4j.md
@@ -30,6 +30,7 @@ the returned fields to a SeaTunnel schema.
| Float | FLOAT / DOUBLE |
| ByteArray | BYTES |
| Date | DATE |
+| LocalTime | TIME |
| LocalDateTime | TIMESTAMP |
| List | ARRAY |
| Map | MAP |
@@ -55,6 +56,8 @@ the returned fields to a SeaTunnel schema.
- Use exactly one authentication method: username/password, bearer token, or
Kerberos ticket.
- `query` controls which fields are returned. `schema.fields` must list the
returned field names and their SeaTunnel types.
- Returned field names can contain dots, such as `t.string`, when the Cypher
query returns properties from a node.
+- `MAP` fields must use `STRING` keys, for example `MAP<STRING, INT>`.
+- Neo4j integer and floating-point values are converted according to the
SeaTunnel type declared in `schema.fields`. Use `BIGINT`/`DOUBLE` when the
value may exceed the range of `INT`/`FLOAT`.
## Task Example
diff --git a/docs/en/connectors/source/Oracle-CDC.md
b/docs/en/connectors/source/Oracle-CDC.md
index 956fccdd3a..31346ed744 100644
--- a/docs/en/connectors/source/Oracle-CDC.md
+++ b/docs/en/connectors/source/Oracle-CDC.md
@@ -241,6 +241,7 @@ exit;
| 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.
[...]
| connection.pool.size | Integer | No | 20 |
The jdbc connection pool size.
[...]
+| incremental.parallelism | Integer | No | 1 |
Number of parallel readers used after the snapshot phase enters incremental log
reading.
[...]
| chunk-key.even-distribution.factor.upper-bound | Double | No | 100
| The upper bound of the chunk key distribution factor. This factor is used
to determine whether the table data is evenly distributed. If the distribution
factor is calculated to be less than or equal to this upper bound (i.e.,
(MAX(id) - MIN(id) + 1) / row count), the table chunks would be optimized for
even distribution. Otherwise, if the distribution factor is greater, the table
will be considered as une [...]
| chunk-key.even-distribution.factor.lower-bound | Double | No | 0.05
| The lower bound of the chunk key distribution factor. This factor is used
to determine whether the table data is evenly distributed. If the distribution
factor is calculated to be greater than or equal to this lower bound (i.e.,
(MAX(id) - MIN(id) + 1) / row count), the table chunks would be optimized for
even distribution. Otherwise, if the distribution factor is less, the table
will be considered as une [...]
| sample-sharding.threshold | Integer | No | 1000 |
This configuration specifies the threshold of estimated shard count to trigger
the sample sharding strategy. When the distribution factor is outside the
bounds specified by `chunk-key.even-distribution.factor.upper-bound` and
`chunk-key.even-distribution.factor.lower-bound`, and the estimated shard count
(calculated as approximate row count / chunk size) exceeds this threshold, the
sample sharding strategy wil [...]
@@ -366,6 +367,35 @@ source {
}
```
+### Enable exactly-once CDC
+
+`exactly_once = true` is used with the default `startup.mode = "initial"`
path. Use it when the downstream sink is also configured for exactly-once
delivery, for example a JDBC sink with XA enabled.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+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"
+ exactly_once = true
+ connection.pool.size = 1
+ debezium {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
+ }
+}
+```
+
### Start From Timestamp
Use `startup.mode = "timestamp"` to start from the Oracle SCN resolved from a
Unix timestamp in milliseconds.
@@ -390,6 +420,29 @@ source {
}
```
+### Configure Debezium heartbeat
+
+For low-traffic tables, Debezium heartbeat can help the Oracle LogMiner
position move forward regularly.
+
+```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"
+ debezium {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ heartbeat.interval.ms = 1000
+ heartbeat.action.query = "INSERT INTO DEBEZIUM.heartbeat (ts) VALUES
(SYSTIMESTAMP)"
+ }
+ }
+}
+```
+
### Schema change event filtering
When `schema-changes.enabled = true`, you can further control which schema
change event types are
diff --git a/docs/zh/connectors/sink/Druid.md b/docs/zh/connectors/sink/Druid.md
index 275cc5878f..3fa9a7dce0 100644
--- a/docs/zh/connectors/sink/Druid.md
+++ b/docs/zh/connectors/sink/Druid.md
@@ -75,6 +75,10 @@ Druid 写入需要主时间列。连接器会自动追加一个名为 `timestamp
该 Sink 适合追加式批量写入，不会把 CDC 的更新/删除行自动转换成 Druid 的 upsert 或 delete 操作。
+仅支持 [数据类型映射](#数据类型映射) 中列出的 SeaTunnel 类型，其他类型会在写入规划阶段报错。
+
+由于连接器会把数据以内联 CSV 的形式提交给 Druid，字符串字段中不建议直接包含英文逗号或换行符；如有这类内容，建议在进入 Druid Sink
前先做清洗或替换。
+
## 示例
### 写入单表
@@ -119,6 +123,7 @@ sink {
Druid {
coordinatorUrl = "router:8888"
datasource = "testDataSource"
+ batchSize = 10000
}
}
```
diff --git a/docs/zh/connectors/sink/Neo4j.md b/docs/zh/connectors/sink/Neo4j.md
index 85f2b09089..dc9fd0fd33 100644
--- a/docs/zh/connectors/sink/Neo4j.md
+++ b/docs/zh/connectors/sink/Neo4j.md
@@ -39,6 +39,8 @@ Neo4j Sink 连接器通过执行 Cypher 语句把 SeaTunnel 数据写入 Neo4j
- `ONE_BY_ONE` 模式下，`queryParamPosition` 用来把 Cypher 占位符映射到输入行的字段位置。
- `BATCH` 模式下，查询语句应使用 `UNWIND $batch AS row`，连接器会通过 `batch` 变量传入一批数据。
- `BATCH` 模式虽然从 `row` 中取值，但连接器配置校验仍要求填写 `queryParamPosition`。
+- `queryParamPosition` 中的字段位置从 `0` 开始，顺序对应上游输入表结构。
+- `BATCH` 模式下，每个 `row` 使用上游字段名取值，因此 Cypher 语句里的字段名需要和上游表结构一致。
## 逐条写入示例
diff --git a/docs/zh/connectors/source/Neo4j.md
b/docs/zh/connectors/source/Neo4j.md
index 35c57d6509..b4d20b7685 100644
--- a/docs/zh/connectors/source/Neo4j.md
+++ b/docs/zh/connectors/source/Neo4j.md
@@ -30,6 +30,7 @@ Neo4j 源连接器通过执行 Cypher 查询从 Neo4j 读取数据，并把查
| Float | FLOAT / DOUBLE |
| ByteArray | BYTES |
| Date | DATE |
+| LocalTime | TIME |
| LocalDateTime | TIMESTAMP |
| List | ARRAY |
| Map | MAP |
@@ -55,6 +56,8 @@ Neo4j 源连接器通过执行 Cypher 查询从 Neo4j 读取数据，并把查
- 认证方式只选一种：用户名密码、bearer token 或 Kerberos ticket。
- `query` 决定返回哪些字段，`schema.fields` 必须写清这些返回字段和对应类型。
- 查询返回字段名可以包含点号，例如从节点属性返回的 `t.string`。
+- `MAP` 字段的 key 必须是 `STRING`，例如 `MAP<STRING, INT>`。
+- Neo4j 的整数和浮点数会按 `schema.fields` 中声明的 SeaTunnel 类型转换；如果数值可能超过 `INT` 或 `FLOAT`
范围，建议使用 `BIGINT` 或 `DOUBLE`。
## 任务示例
diff --git a/docs/zh/connectors/source/Oracle-CDC.md
b/docs/zh/connectors/source/Oracle-CDC.md
index e87827059e..1efec00fe9 100644
--- a/docs/zh/connectors/source/Oracle-CDC.md
+++ b/docs/zh/connectors/source/Oracle-CDC.md
@@ -240,6 +240,7 @@ exit;
| connect.timeout.ms | Duration | 否 | 30000 |
连接器在尝试连接数据库服务器后超时的最大等待时间。
[...]
| connect.max-retries | Integer | 否 | 3 |
连接器尝试建立数据库服务器连接的最大重试次数。
[...]
| connection.pool.size | Integer | 否 | 20 |
JDBC 连接池大小。
[...]
+| incremental.parallelism | Integer | 否 | 1 |
全量快照阶段结束、进入增量日志读取后使用的并行读取数量。
[...]
| chunk-key.even-distribution.factor.upper-bound | Double | 否 | 100
| 分块键分布因子的上限。此因子用于确定表数据是否均匀分布。如果计算出的分布因子小于或等于此上限（即 (MAX(id) - MIN(id) + 1) /
行数），则表分块将针对均匀分布进行优化。否则，如果分布因子较大，则表将被视为分布不均，如果估计的分片数超过
`sample-sharding.threshold` 指定的值，则将使用基于采样的分片策略。默认值为 100.0。 |
| chunk-key.even-distribution.factor.lower-bound | Double | 否 | 0.05
| 分块键分布因子的下限。此因子用于确定表数据是否均匀分布。如果计算出的分布因子大于或等于此下限（即 (MAX(id) - MIN(id) + 1) /
行数），则表分块将针对均匀分布进行优化。否则，如果分布因子较小，则表将被视为分布不均，如果估计的分片数超过
`sample-sharding.threshold` 指定的值，则将使用基于采样的分片策略。默认值为 0.05。 |
| sample-sharding.threshold | Integer | 否 | 1000 |
此配置指定触发采样分片策略的预估分片数阈值。当分布因子超出 `chunk-key.even-distribution.factor.upper-bound`
和 `chunk-key.even-distribution.factor.lower-bound` 指定的范围，并且预估的分片数（计算为近似行数 /
分块大小）超过此阈值时，将使用采样分片策略。这有助于更有效地处理大型数据集。默认值为 1000 个分片。
|
@@ -365,6 +366,35 @@ source {
}
```
+### 启用精确一次 CDC
+
+`exactly_once = true` 用于默认的 `startup.mode = "initial"` 路径。只有下游 Sink
也配置了精确一次能力时才建议启用，例如开启 XA 的 JDBC Sink。
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+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"
+ exactly_once = true
+ connection.pool.size = 1
+ debezium {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
+ }
+}
+```
+
### 从时间戳启动
使用 `startup.mode = "timestamp"` 时，Oracle CDC 会根据毫秒级 Unix 时间戳解析对应的 Oracle SCN
并从该位置启动。
@@ -389,6 +419,29 @@ source {
}
```
+### 配置 Debezium 心跳
+
+对于变更较少的表，可以配置 Debezium 心跳，让 Oracle LogMiner 位点定期向前推进。
+
+```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"
+ debezium {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ heartbeat.interval.ms = 1000
+ heartbeat.action.query = "INSERT INTO DEBEZIUM.heartbeat (ts) VALUES
(SYSTIMESTAMP)"
+ }
+ }
+}
+```
+
### Schema change 事件过滤
当 `schema-changes.enabled = true` 时，可通过 `schema-changes.include` /
`schema-changes.exclude` 进一步
