This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 84e43a95b4 [Docs][Connector-V2] Improve InfluxDB Greenplum Mqtt and
Aerospike connector docs (#11722)
84e43a95b4 is described below
commit 84e43a95b469b2021b0590e0bbac703bcc9cdcf3
Author: Daniel Carter <[email protected]>
AuthorDate: Mon Sep 7 05:47:37 2026 +0000
[Docs][Connector-V2] Improve InfluxDB Greenplum Mqtt and Aerospike
connector docs (#11722)
Co-authored-by: DanielCarter-stack
<[email protected]>
---
docs/en/connectors/sink/Aerospike.md | 42 ++++++++++++++++++++++
docs/en/connectors/sink/Greenplum.md | 8 ++---
docs/en/connectors/sink/InfluxDB.md | 37 +++++++++++++++++++
docs/en/connectors/source/Greenplum.md | 14 ++++----
docs/en/connectors/source/InfluxDB.md | 66 ++++++++++++++++++++++++++++++++++
docs/en/connectors/source/Mqtt.md | 34 +++++++++---------
docs/zh/connectors/source/Greenplum.md | 8 ++---
7 files changed, 177 insertions(+), 32 deletions(-)
diff --git a/docs/en/connectors/sink/Aerospike.md
b/docs/en/connectors/sink/Aerospike.md
index 6442da5356..a1f25836d4 100644
--- a/docs/en/connectors/sink/Aerospike.md
+++ b/docs/en/connectors/sink/Aerospike.md
@@ -154,6 +154,48 @@ sink {
}
```
+### Stream CDC Records Into Aerospike With `kv` Format
+
+For streaming jobs (for example MySQL CDC), use `data_format = "kv"` to write
each configured
+field as an independent Aerospike bin. This makes downstream lookups cheaper
because each field
+is stored separately instead of being packed into a single JSON or map bin.
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "STREAMING"
+ checkpoint.interval = 10000
+}
+
+source {
+ Mysql-CDC {
+ url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
+ username = "root"
+ password = "******"
+ table-names = ["seatunnel.user"]
+ }
+}
+
+sink {
+ Aerospike {
+ host = "aerospike-host"
+ port = 3000
+ namespace = "test"
+ set = "user"
+ key = "id"
+ data_format = "kv"
+ write_timeout = 500
+ schema {
+ field {
+ id = "INTEGER"
+ name = "STRING"
+ email = "STRING"
+ }
+ }
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/sink/Greenplum.md
b/docs/en/connectors/sink/Greenplum.md
index e87ca40f30..374f648e67 100644
--- a/docs/en/connectors/sink/Greenplum.md
+++ b/docs/en/connectors/sink/Greenplum.md
@@ -51,14 +51,14 @@ Only Greenplum-specific commonly used options are listed
here. Other JDBC sink o
| Name | Type | Required | Default | Description |
|------|------|----------|---------|-------------|
-| url | String | Yes | - | JDBC connection URL. Use
`jdbc:postgresql://host:port/database` with PostgreSQL driver, or
`jdbc:pivotal:greenplum://host:port;DatabaseName=database` with Greenplum
native driver. |
+| url | String | Yes | - | JDBC connection URL. Use
`jdbc:postgresql://host:port/database` with the PostgreSQL driver, or
`jdbc:pivotal:greenplum://host:port;DatabaseName=database` with the Greenplum
native driver. |
| driver | String | Yes | - | JDBC driver class name, usually
`org.postgresql.Driver` or `com.pivotal.jdbc.GreenplumDriver`. |
| username | String | No | - | Greenplum username. |
| password | String | No | - | Greenplum password. |
-| query | String | No | - | SQL used to write upstream rows, for example
`insert into sink(age, name) values(?, ?)`. `query` has higher priority than
generated sink SQL. |
-| batch_size | Int | No | 1000 | Maximum records buffered before flushing to
Greenplum. |
+| query | String | No | - | SQL used to write upstream rows, for example
`insert into sink(age, name) values(?, ?)`. `query` has higher priority than
generated sink SQL. When unset, the sink expects `generate_sink_sql = true`
together with `database` and `table`. |
+| batch_size | Int | No | 1000 | Maximum records buffered before flushing to
Greenplum. The buffer is also flushed at each checkpoint. |
| max_retries | Int | No | 0 | Retry times after `executeBatch` fails. |
-| generate_sink_sql | Boolean | No | false | Generate insert SQL automatically
from `database` and `table`. |
+| generate_sink_sql | Boolean | No | false | Generate the insert SQL
automatically from `database` and `table`. When `true`, the column order must
match the upstream schema. |
| database | String | No | - | Database name used when `generate_sink_sql =
true`. |
| table | String | No | - | Target table name used when `generate_sink_sql =
true`. |
| common-options | | No | - | Sink plugin common parameters, please refer to
[Sink Common Options](../common-options/sink-common-options.md) for details. |
diff --git a/docs/en/connectors/sink/InfluxDB.md
b/docs/en/connectors/sink/InfluxDB.md
index cabe7cdeb6..f849eee7ca 100644
--- a/docs/en/connectors/sink/InfluxDB.md
+++ b/docs/en/connectors/sink/InfluxDB.md
@@ -239,6 +239,43 @@ sink {
}
```
+### Per-Table Measurement With Replicated Writers
+
+Use `multi_table_sink_replica` to spread multi-table writes across more writer
instances when
+the upstream emits a large number of tables.
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "STREAMING"
+ checkpoint.interval = 10000
+}
+
+source {
+ Mysql-CDC {
+ url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
+ username = "root"
+ password = "******"
+ table-names = ["seatunnel.role", "seatunnel.user", "galileo.Bucket"]
+ }
+}
+
+sink {
+ InfluxDB {
+ url = "http://influxdb-host:8086";
+ database = "test"
+ key_time = "time"
+ key_tags = ["label"]
+ batch_size = 2048
+ max_retries = 3
+ retry_backoff_multiplier_ms = 100
+ max_retry_backoff_ms = 5000
+ write_timeout = 10
+ multi_table_sink_replica = 2
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/source/Greenplum.md
b/docs/en/connectors/source/Greenplum.md
index 15557903ca..beee23ce15 100644
--- a/docs/en/connectors/source/Greenplum.md
+++ b/docs/en/connectors/source/Greenplum.md
@@ -56,16 +56,16 @@ Only Greenplum-specific commonly used options are listed
here. Other JDBC source
| Name | Type | Required | Default | Description |
|------|------|----------|---------|-------------|
-| url | String | Yes | - | JDBC connection URL. Use
`jdbc:postgresql://host:port/database` with PostgreSQL driver, or
`jdbc:pivotal:greenplum://host:port;DatabaseName=database` with Greenplum
native driver. |
+| url | String | Yes | - | JDBC connection URL. Use
`jdbc:postgresql://host:port/database` with the PostgreSQL driver, or
`jdbc:pivotal:greenplum://host:port;DatabaseName=database` with the Greenplum
native driver. |
| driver | String | Yes | - | JDBC driver class name, usually
`org.postgresql.Driver` or `com.pivotal.jdbc.GreenplumDriver`. |
| username | String | No | - | Greenplum username. |
| password | String | No | - | Greenplum password. |
-| query | String | Yes | - | SQL used to read data. You can select only the
columns you need. |
-| partition_column | String | No | - | Column used to split data for parallel
reading. |
-| partition_lower_bound | Long | No | - | Lower bound of `partition_column`.
If not set, SeaTunnel queries the minimum value. |
-| partition_upper_bound | Long | No | - | Upper bound of `partition_column`.
If not set, SeaTunnel queries the maximum value. |
-| partition_num | Int | No | job parallelism | Number of source splits. |
-| split.string_split_mode | String | No | sample | String split algorithm. Use
`charset_based` when the split column contains printable ASCII strings and you
want deterministic range-like splitting. |
+| query | String | Yes | - | SQL used to read data. Select only the columns
you need; the source uses the SELECT column list as the output schema. |
+| partition_column | String | No | - | Column used to split data for parallel
reading. Can be numeric or string. Numeric columns are split into
`partition_num` numeric ranges; string columns are split by hash
(`split.string_split_mode = sample`, the default) or by lexicographic range
(`split.string_split_mode = charset_based`). |
+| partition_lower_bound | String | No | - | Lower bound of `partition_column`.
If not set, SeaTunnel queries the minimum value. |
+| partition_upper_bound | String | No | - | Upper bound of `partition_column`.
If not set, SeaTunnel queries the maximum value. |
+| partition_num | Int | No | 10 | Number of source splits. Each split issues a
range query for numeric (and `charset_based` string) columns, or a hash-modulo
predicate for `sample` string columns. Defaults to `10` when unset; tune upward
to match the job's parallelism if needed. |
+| split.string_split_mode | String | No | sample | String split algorithm.
`sample` estimates splits from a sampled value list; `charset_based` performs
deterministic range-like splitting and is preferred when the split column
contains printable ASCII strings. |
| common-options | | No | - | Source plugin common parameters, please refer to
[Source Common Options](../common-options/source-common-options.md) for
details. |
### Tips
diff --git a/docs/en/connectors/source/InfluxDB.md
b/docs/en/connectors/source/InfluxDB.md
index 2ad1633331..c3d6581ce1 100644
--- a/docs/en/connectors/source/InfluxDB.md
+++ b/docs/en/connectors/source/InfluxDB.md
@@ -140,6 +140,21 @@ Number of query splits. Configure it together with
`lower_bound`, `upper_bound`,
`split_column`. Make sure `upper_bound - lower_bound` is divisible by
`partition_num`; otherwise
the query results overlap.
+### where [string]
+
+:::caution Reserved option
+
+The current split logic reads the lowercase `where` keyword directly from the
`sql` option.
+Setting `where` has no effect on the generated split queries — put the filter
inside `sql`
+instead, for example `select * from test where age > 0`. The split parser is
case-sensitive on
+the `where` keyword.
+
+This option is kept in the option list only because the validation rule
+(`InfluxDBSourceFactory.optionRule()`) still references it for backward
compatibility; the
+runtime split query generator does not read it.
+
+:::
+
### epoch [string]
Time precision returned by InfluxDB. Valid values include `H`, `m`, `s`, `MS`,
`u`, and `n`. The
@@ -265,6 +280,57 @@ sink {
}
```
+### Bounded Read With Parallel Splits
+
+The InfluxDB source is bounded by the `sql` result. To chain it to a streaming
sink without
+losing rows, set a finite `partition_num` and let SeaTunnel checkpoint the
offsets per split.
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "BATCH"
+ checkpoint.interval = 10000
+}
+
+source {
+ InfluxDB {
+ url = "http://influxdb-host:8086";
+ sql = "select label, c_string, c_double, c_bigint, c_float, c_int,
c_smallint, c_boolean from source"
+ database = "test"
+ lower_bound = 0
+ upper_bound = 99
+ partition_num = 4
+ split_column = "c_int"
+ query_timeout_sec = 10
+ connect_timeout_ms = 20000
+ schema {
+ fields {
+ label = STRING
+ c_string = STRING
+ c_double = DOUBLE
+ c_bigint = BIGINT
+ c_float = FLOAT
+ c_int = INT
+ c_smallint = SMALLINT
+ c_boolean = BOOLEAN
+ time = BIGINT
+ }
+ }
+ }
+}
+
+sink {
+ InfluxDB {
+ url = "http://influxdb-host:8086";
+ database = "test"
+ measurement = "sink"
+ key_time = "time"
+ key_tags = ["label"]
+ batch_size = 1024
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/source/Mqtt.md
b/docs/en/connectors/source/Mqtt.md
index 0a1faf9129..e82e7edeb1 100644
--- a/docs/en/connectors/source/Mqtt.md
+++ b/docs/en/connectors/source/Mqtt.md
@@ -37,23 +37,23 @@ The source uses MQTT auto-reconnect. If the client remains
disconnected longer t
## Options
-| name | type | required | default value |
-|---------------------|---------|----------|---------------|
-| url | string | yes | - |
-| topic | string | yes | - |
-| schema | config | yes | - |
-| username | string | no | - |
-| password | string | no | - |
-| qos | int | no | 1 |
-| format | string | no | json |
-| field_delimiter | string | no | , |
-| client_id | string | no | - |
-| clean_session | boolean | no | true |
-| connection_timeout | int | no | 30 |
-| keep_alive_interval | int | no | 60 |
-| reconnect_timeout | int | no | 120 |
-| max_queue_size | int | no | 1000 |
-| common-options | | no | - |
+| name | type | required | default value | description
|
+|---------------------|---------|----------|---------------|------------------------------------------------------------------------------------------------------------------------|
+| url | string | yes | - | MQTT broker URL,
e.g. `tcp://localhost:1883`.
|
+| topic | string | yes | - | MQTT topic to
subscribe messages from.
|
+| schema | config | yes | - | Output schema of
the source. See [Schema
Feature](../../introduction/concepts/schema-feature.md). |
+| username | string | no | - | MQTT broker
authentication username. Leave unset for anonymous access.
|
+| password | string | no | - | MQTT broker
authentication password. Leave unset for anonymous access.
|
+| qos | int | no | 1 | MQTT subscribe
QoS level. `0` is at-most-once, `1` is at-least-once.
|
+| format | string | no | json | Message
deserialization format: `json` or `text`.
|
+| field_delimiter | string | no | , | Field delimiter
for `text` format. Only used when `format=text`.
|
+| client_id | string | no | - | MQTT client id.
Required when `clean_session=false`; generated automatically otherwise.
|
+| clean_session | boolean | no | true | Whether to use a
clean MQTT session.
|
+| connection_timeout | int | no | 30 | MQTT connection
timeout in seconds.
|
+| keep_alive_interval | int | no | 60 | MQTT keep alive
interval in seconds.
|
+| reconnect_timeout | int | no | 120 | Maximum seconds
to wait for MQTT auto-reconnect before failing the source.
|
+| max_queue_size | int | no | 1000 | Maximum number of
MQTT messages buffered in memory before deserialization.
|
+| common-options | | no | - | Source plugin
common parameters. See [Source Common
Options](../common-options/source-common-options.md) for details. |
### url [string]
diff --git a/docs/zh/connectors/source/Greenplum.md
b/docs/zh/connectors/source/Greenplum.md
index 93e1fb3cd5..feefaa9fd7 100644
--- a/docs/zh/connectors/source/Greenplum.md
+++ b/docs/zh/connectors/source/Greenplum.md
@@ -61,10 +61,10 @@ import ChangeLog from '../changelog/connector-jdbc.md';
| username | String | 否 | - | Greenplum 用户名。 |
| password | String | 否 | - | Greenplum 密码。 |
| query | String | 是 | - | 读取数据的 SQL，可以只选择需要的字段。 |
-| partition_column | String | 否 | - | 用于并行拆分读取的字段。 |
-| partition_lower_bound | Long | 否 | - | `partition_column` 的下界；不配置时 SeaTunnel
会查询最小值。 |
-| partition_upper_bound | Long | 否 | - | `partition_column` 的上界；不配置时 SeaTunnel
会查询最大值。 |
-| partition_num | Int | 否 | 作业并行度 | 拆分数量。 |
+| partition_column | String | 否 | - | 用于并行拆分读取的字段，可以是数值或字符串类型。数值列按
`partition_num` 切分为数值范围；字符串列在 `split.string_split_mode = sample`（默认）下按哈希拆分，在
`charset_based` 下按字典序范围拆分。 |
+| partition_lower_bound | String | 否 | - | `partition_column` 的下界；不配置时
SeaTunnel 会查询最小值。 |
+| partition_upper_bound | String | 否 | - | `partition_column` 的上界；不配置时
SeaTunnel 会查询最大值。 |
+| partition_num | Int | 否 | 10 | 拆分数量。数值列（以及 `charset_based`
字符串列）每个拆分使用范围查询；`sample` 字符串列每个拆分使用哈希取模谓词。未设置时默认为 `10`，可按需上调以匹配作业并行度。 |
| split.string_split_mode | String | 否 | sample | 字符串拆分算法。拆分字段是可打印 ASCII
字符串，并且希望使用确定性的范围类拆分时，可以配置为 `charset_based`。 |
| common-options | | 否 | - | 源插件通用参数，请参考
[源通用选项](../common-options/source-common-options.md)。 |
