This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-11722-85ed97f4c0a53f881b9d0d5a454b4fdb885d98e9 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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)。 |
