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-11694-886a3ac6e59aad5e46755761cf9fb8ac2a6d3df9 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit fb47236179b271bf279f3777734c0e3b3e9287db Author: Daniel Carter <[email protected]> AuthorDate: Thu Oct 29 09:56:33 2026 +0000 [Docs][Connector-V2] Improve Clickhouse Http MongoDB Redis and Easysearch connector docs (#11694) Co-authored-by: DanielCarter-stack <[email protected]> Co-authored-by: davidzollo <[email protected]> --- docs/en/connectors/sink/Clickhouse.md | 45 ++++++++++++++++++ docs/en/connectors/sink/Easysearch.md | 84 +++++++++++++++++++++++++-------- docs/en/connectors/sink/MongoDB.md | 60 +++++++++++++++++++++++ docs/en/connectors/source/Clickhouse.md | 58 ++++++++++++++++++----- docs/en/connectors/source/Easysearch.md | 1 + docs/en/connectors/source/Http.md | 71 ++++++++++++++++++++++++++++ docs/en/connectors/source/MongoDB.md | 49 +++++++++++++++++++ docs/en/connectors/source/Redis.md | 28 +++++++++++ docs/zh/connectors/sink/Clickhouse.md | 44 +++++++++++++++++ docs/zh/connectors/sink/Easysearch.md | 83 ++++++++++++++++++++++++-------- docs/zh/connectors/sink/MongoDB.md | 60 +++++++++++++++++++++++ docs/zh/connectors/sink/Redis.md | 18 +++++++ docs/zh/connectors/source/Clickhouse.md | 54 ++++++++++++++++----- docs/zh/connectors/source/Easysearch.md | 1 + docs/zh/connectors/source/Http.md | 67 ++++++++++++++++++++++++++ docs/zh/connectors/source/MongoDB.md | 48 +++++++++++++++++++ docs/zh/connectors/source/Redis.md | 28 +++++++++++ 17 files changed, 737 insertions(+), 62 deletions(-) diff --git a/docs/en/connectors/sink/Clickhouse.md b/docs/en/connectors/sink/Clickhouse.md index 833c4e2785..3196726eb6 100644 --- a/docs/en/connectors/sink/Clickhouse.md +++ b/docs/en/connectors/sink/Clickhouse.md @@ -318,6 +318,51 @@ sink { } ``` +### CDC Changelog Sink + +When the upstream is a changelog source (for example MySQL CDC or Postgres CDC), the rows contain `INSERT`, +`UPDATE_BEFORE`/`UPDATE_AFTER`, and `DELETE` events. Configure `primary_key` so the sink can map `UPDATE` and +`DELETE` rows to the right target row, and pair with `support_upsert = true` if the target table does not already +provide a deduplication engine. + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + schema = { + fields { + pk_id = bigint + name = string + score = int + } + } + rows = [ + { kind = INSERT, fields = [1, "A", 100] } + { kind = INSERT, fields = [2, "B", 100] } + { kind = INSERT, fields = [3, "C", 100] } + { kind = UPDATE_BEFORE, fields = [1, "A", 100] } + { kind = UPDATE_AFTER, fields = [1, "A_1", 100] } + { kind = DELETE, fields = [2, "B", 100] } + ] + } +} + +sink { + Clickhouse { + host = "localhost:8123" + database = "default" + table = "sink_table" + username = "default" + password = "" + primary_key = "pk_id" + } +} +``` + ### Multiple table Sink Cases In ClickHouse, create the following two data tables in advance: diff --git a/docs/en/connectors/sink/Easysearch.md b/docs/en/connectors/sink/Easysearch.md index 15071472df..894c3bf331 100644 --- a/docs/en/connectors/sink/Easysearch.md +++ b/docs/en/connectors/sink/Easysearch.md @@ -21,6 +21,7 @@ A sink plugin used to send data to `INFINI Easysearch`. - [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md) - [x] [cdc](../../introduction/concepts/connector-v2-features.md) - [x] [batch](../../introduction/concepts/connector-v2-features.md) +- [ ] [support multiple table write](../../introduction/concepts/connector-v2-features.md) :::tip @@ -46,26 +47,26 @@ Engine Supported ## Sink Options -| name | type | required | default value | -|------------------------|---------|----------|---------------| -| hosts | array | yes | - | -| index | string | yes | - | -| primary_keys | list | no | | -| key_delimiter | string | no | `_` | -| key_encoding | enum | no | LEGACY | -| username | string | no | | -| password | string | no | | -| max_retry_count | int | no | 3 | -| max_batch_size | int | no | 10 | -| tls_verify_certificate | boolean | no | true | -| tls_verify_hostname | boolean | no | true | -| tls_keystore_path | string | no | - | -| tls_keystore_password | string | no | - | -| tls_truststore_path | string | no | - | -| tls_truststore_password | string | no | - | -| schema_save_mode | enum | no | CREATE_SCHEMA_WHEN_NOT_EXIST | -| data_save_mode | enum | no | APPEND_DATA | -| common-options | | no | - | +| name | type | required | default value | description | +|------------------------|---------|----------|---------------|-------------------------------------------------------------------------------------------------------------------| +| hosts | array | yes | - | Easysearch HTTP cluster addresses in `host:port` format, for example `["host1:9200", "host2:9200"]`. | +| index | string | yes | - | Easysearch index name. Supports placeholders such as `seatunnel_${age}`. | +| primary_keys | list | no | - | Primary key fields used to build the document `_id`. Configure when writing CDC rows that need upsert/delete. | +| key_delimiter | string | no | `_` | Delimiter used to join composite key fields when building `_id`. | +| key_encoding | enum | no | LEGACY | Encoding used to build document keys; see `key_encoding` below. | +| username | string | no | - | Username for secured Easysearch clusters. | +| password | string | no | - | Password for secured Easysearch clusters. | +| max_retry_count | int | no | 3 | Maximum retry count for one bulk request. | +| max_batch_size | int | no | 10 | Maximum number of documents buffered into one bulk request. | +| tls_verify_certificate | boolean | no | true | Whether to validate HTTPS certificates. | +| tls_verify_hostname | boolean | no | true | Whether to validate HTTPS host names. | +| tls_keystore_path | string | no | - | Path to the PEM or JKS key store. | +| tls_keystore_password | string | no | - | Password for the configured key store. | +| tls_truststore_path | string | no | - | Path to the PEM or JKS trust store. | +| tls_truststore_password | string | no | - | Password for the configured trust store. | +| schema_save_mode | enum | no | CREATE_SCHEMA_WHEN_NOT_EXIST | How to handle the target index before the synchronization task starts. See `schema_save_mode` below. | +| data_save_mode | enum | no | APPEND_DATA | How to handle existing target data before the synchronization task starts. See `data_save_mode` below. | +| common-options | | no | - | Sink plugin common parameters, please refer to [Sink Common Options](../common-options/sink-common-options.md). | ### hosts [array] @@ -185,6 +186,49 @@ sink { } ``` +### Route Rows To Dynamic Indexes + +Easysearch does not implement the multi-table sink contract. This example uses one +input table with a `target_index` field on each row. The index template reads that +field and writes the two rows to `st_customers` and `st_orders`. Upstream table +identifiers are not automatically exposed as row fields. `target_index` is a +row field; the reserved `${table_name}` metadata placeholder is not used here. +Create the target indexes first or enable index auto-creation on the server. +`schema_save_mode = "IGNORE"` leaves their schema management to the server. + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + row.num = 2 + schema { + fields { + id = int + name = string + target_index = string + } + } + rows = [ + { kind = INSERT, fields = [1, "alice", "customers"] }, + { kind = INSERT, fields = [2, "order", "orders"] } + ] + } +} + +sink { + Easysearch { + hosts = ["localhost:9200"] + index = "st_${target_index}" + primary_keys = ["id"] + schema_save_mode = "IGNORE" + } +} +``` + SSL (Disable certificates validation) ```hocon diff --git a/docs/en/connectors/sink/MongoDB.md b/docs/en/connectors/sink/MongoDB.md index a7e3fa5cc4..c3d39a2555 100644 --- a/docs/en/connectors/sink/MongoDB.md +++ b/docs/en/connectors/sink/MongoDB.md @@ -330,6 +330,66 @@ sink { } ``` +### CDC Changelog Sink + +When the upstream is a changelog source, configure `upsert-enable = true` and `primary-key` so `UPDATE` and +`DELETE` events map to existing documents instead of being inserted as duplicates. The sink also exposes +`buffer-flush.max-rows`, `buffer-flush.interval`, and `retry.max` to control the batch flush and retry behaviour. + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + schema = { + fields { + id = int + name = string + } + } + rows = [ + { kind = INSERT, fields = [1, "alice"] } + { kind = INSERT, fields = [2, "bob"] } + { kind = UPDATE_BEFORE, fields = [1, "alice"] } + { kind = UPDATE_AFTER, fields = [1, "alice_1"] } + { kind = DELETE, fields = [2, "bob"] } + ] + } +} + +sink { + MongoDB { + uri = "mongodb://user:[email protected]:27017/test_db?retryWrites=true" + database = "test_db" + collection = "users" + upsert-enable = true + primary-key = ["id"] + buffer-flush.max-rows = 2000 + buffer-flush.interval = 1000 + } +} +``` + +### Transactional Sink (MongoDB 4.2+) + +Set `transaction = true` to wrap the bulk writes into a MongoDB multi-document transaction. This option requires +MongoDB 4.2 or later and is committed at checkpoint boundaries. Note that timer flush is automatically disabled +in transaction mode so that the checkpoint-driven commit boundary is preserved. + +```hocon +sink { + MongoDB { + uri = "mongodb://user:[email protected]:27017/test_db?replicaSet=rs0" + database = "test_db" + collection = "orders" + transaction = true + } +} +``` + ## Changelog <ChangeLog /> diff --git a/docs/en/connectors/source/Clickhouse.md b/docs/en/connectors/source/Clickhouse.md index f15ce69bca..1ebc1766f7 100644 --- a/docs/en/connectors/source/Clickhouse.md +++ b/docs/en/connectors/source/Clickhouse.md @@ -197,26 +197,30 @@ sink { ``` ### Multiple table + +When `table_list` contains multiple entries, SeaTunnel emits one row stream per upstream table and uses `${table_name}` in +the downstream connector to route each stream to the matching target. Combine this with the Clickhouse sink's +`table = "${table_name}_multi_table_sink"` style configuration to land each upstream table in its own target table. + ```hocon env { job.mode = "BATCH" - parallelism = 5 + parallelism = 3 } source { Clickhouse { host = "localhost:8123" - username = "xxx" - password = "xxx" + username = "default" + password = "" table_list = [ { - table_path = "default.table1" - sql = "select * from default.table1 where id > 2 and type = 1" + table_path = "default.source_table" + sql = "select * from source_table" }, { - table_path = "default.table2" - sql = "select * from default.table2 where age > 18" - split_size = 1 + table_path = "default.source_merge_tree_table" + filter_query = "id < 47" } ] server_time_zone = "UTC" @@ -226,14 +230,46 @@ source { } } -# Console printing of the read Clickhouse data +# Each upstream table is written to a separate target table. sink { - Console { - parallelism = 1 + Clickhouse { + host = "localhost:8123" + database = "default" + table = "${table_name}_multi_table_sink" + username = "default" + password = "" } } ``` +## FAQ + +### How do I read only some partitions or rows of a ClickHouse table? + +Prefer `partition_list` to filter partitions or `filter_query` to apply a server-side predicate. Both avoid full-table +scans when the upstream table is large. Combine them with `sql` if you also need to project specific columns; in that +case the connector runs in SQL mode and uses the partition/filter hints alongside the user query. + +### When should I use SQL mode vs query table mode? + +Use SQL mode (`sql` is set) when you need joins, aggregations, subqueries, or projections on multiple tables. Use +query table mode (`table_path` only) when you want to read one logical table with parallel `system.parts`-based +splits and the lowest read latency. The connector falls back to single parallelism for any SQL that needs to run +on a single shard, even when a higher `parallelism` is configured. + +### How do I tune read throughput and memory pressure? + +The two main knobs are `split.size` (how many ClickHouse parts are grouped into one SeaTunnel split) and `batch_size` +(how many rows are pulled per round-trip). Decrease `split.size` to raise parallelism on small part counts, and +increase `batch_size` to amortise round-trip cost on larger parts. The `clickhouse.config` map exposes every +`clickhouse-jdbc` option for further tuning, for example `socket_timeout`. + +### How are `table_list` and a single `table_path` related? + +When the configuration has exactly one logical table, you can flatten the table-list entry to the outer level and +configure `split.size` (note the dot) instead of `split_size`. With multiple entries in `table_list`, keep each +table's options inside its own entry and keep `split_size` inside the entry. + ## Changelog <ChangeLog /> diff --git a/docs/en/connectors/source/Easysearch.md b/docs/en/connectors/source/Easysearch.md index 2ce6e1614d..bb7114b654 100644 --- a/docs/en/connectors/source/Easysearch.md +++ b/docs/en/connectors/source/Easysearch.md @@ -26,6 +26,7 @@ Used to read data from INFINI Easysearch. - [x] [column projection](../../introduction/concepts/connector-v2-features.md) - [ ] [parallelism](../../introduction/concepts/connector-v2-features.md) - [ ] [support user-defined split](../../introduction/concepts/connector-v2-features.md) +- [ ] [support multiple table read](../../introduction/concepts/connector-v2-features.md) :::tip diff --git a/docs/en/connectors/source/Http.md b/docs/en/connectors/source/Http.md index 7793608638..c716409916 100644 --- a/docs/en/connectors/source/Http.md +++ b/docs/en/connectors/source/Http.md @@ -723,6 +723,77 @@ source { ``` +## Streaming Mode + +In `STREAMING` jobs, the connector polls the configured URL on a fixed interval so the source behaves like a long +running pull job. `poll_interval_millis` controls the interval between requests; `retry`, `retry_backoff_multiplier_ms`, +and `retry_backoff_max_ms` together tune the retry behaviour on `IOException`. + +```hocon +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 60000 +} + +source { + Http { + url = "https://api.example.com/events"; + method = "GET" + format = "json" + poll_interval_millis = 5000 + retry = 3 + retry_backoff_multiplier_ms = 200 + retry_backoff_max_ms = 5000 + schema = { + fields { + id = string + type = string + timestamp = string + } + } + } +} + +sink { + Console {} +} +``` + +## FAQ + +### How do I download a binary file (PDF, image, ZIP) through Http? + +Set `format = binary`. The response body is split into chunks of `binary_chunk_size` bytes (default 10 MiB) and +emitted as `(data: bytes, relativePath: string, partIndex: long)` rows. Combine this with a `LocalFile` sink in +BATCH mode to write the chunks to disk. + +### How do I keep the page number as a URL parameter? + +Set `keep_page_param_as_http_param = true`. The connector copies the paging field into `params` so it travels as +a query string parameter on every request. When `false`, the connector only updates existing keys or `${page}` +placeholders in `headers`, `params`, or `body`. + +### What does `use_placeholder_replacement` actually do? + +When `true`, the connector recognises `${page}` and `${cursor}` placeholders in headers, params, and body, and +replaces them with the actual page/cursor value. Prefixed or suffixed forms such as `"10${page}"` (becomes +`"105"` when `page=5`) are also supported. When `false`, only key-based replacement is applied: the paging field +must already exist as a key in the target map. + +### Why does the request body look empty when `method = POST` and I did not set `body`? + +With `keep_params_as_form = false`, the default non-form branch serialises `body` as JSON. When no `body` is +configured the connector still sends a JSON object `{}` as the request body, so HTTP servers see a well-formed +JSON payload. Switch to `keep_params_as_form = true` (and optionally set `Content-Type: application/x-www-form-urlencoded`) +to use the form-body branch instead. + +### How do I limit retries and slow down reconnect attempts? + +Use `retry` for the maximum retry count on `IOException`. `retry_backoff_multiplier_ms` (default `100`) is the +base backoff in milliseconds; `retry_backoff_max_ms` (default `10000`) caps the backoff so a misconfigured value +does not stall the job. + ## Changelog <ChangeLog /> diff --git a/docs/en/connectors/source/MongoDB.md b/docs/en/connectors/source/MongoDB.md index 0bb83beda1..a6408ec424 100644 --- a/docs/en/connectors/source/MongoDB.md +++ b/docs/en/connectors/source/MongoDB.md @@ -387,6 +387,55 @@ A sample document that flows through this configuration looks like the following } ``` +### Read With Partition Split Key + +To accelerate reads on large collections, MongoDB source lets you split the scan by a key (`partition.split-key`) +and a target split size (`partition.split-size`). The connector reads documents ordered by the split key and +hands each chunk to a parallel reader. `fetch.size` controls how many documents the server returns per round-trip +and should be tuned together with the split size. + +```hocon +source { + MongoDB { + uri = "mongodb://user:[email protected]:27017" + database = "test_db" + collection = "orders" + partition.split-key = "_id" + partition.split-size = 1024 + fetch.size = 2048 + max.time-min = 30 + schema = { + fields { + _id = string + status = string + amount = double + } + } + } +} +``` + +### Read From Replica Set Or Sharded Cluster + +The connector accepts any standard MongoDB connection URI. Use `replicaSet=xxx` for replica sets and +`authSource=admin` when credentials live in a non-default auth database. + +```hocon +source { + MongoDB { + uri = "mongodb://user:password@mongo1:27017,mongo2:27017,mongo3:27017/test_db?replicaSet=rs0&authSource=admin&readPreference=secondary" + database = "test_db" + collection = "orders" + schema = { + fields { + _id = string + status = string + } + } + } +} +``` + ## Changelog <ChangeLog /> diff --git a/docs/en/connectors/source/Redis.md b/docs/en/connectors/source/Redis.md index 2b1d987774..ed81dfd978 100644 --- a/docs/en/connectors/source/Redis.md +++ b/docs/en/connectors/source/Redis.md @@ -591,6 +591,34 @@ sink { Console {} } ``` +### Read With Custom Key And Value Field Names + +Use `key_field_name`, `single_field_name`, and `read_key_enabled = true` to include the Redis key in the output and +control how its value maps to a schema column. For primitive value types (`string`, `list`, `set`, `zset`) you +must set `single_field_name` so the connector knows which schema column receives the value. + +```hocon +source { + Redis { + host = "localhost" + port = 6379 + keys = "string_test*" + data_type = string + read_key_enabled = true + key_field_name = custom_key + single_field_name = custom_value + format = json + schema = { + table = "RedisDatabase.RedisTable" + columns = [ + { name = "custom_key", type = "string" } + { name = "custom_value", type = "string" } + ] + } + } +} +``` + ## Changelog <ChangeLog /> diff --git a/docs/zh/connectors/sink/Clickhouse.md b/docs/zh/connectors/sink/Clickhouse.md index f529cbfbbf..eeccde2462 100644 --- a/docs/zh/connectors/sink/Clickhouse.md +++ b/docs/zh/connectors/sink/Clickhouse.md @@ -313,6 +313,50 @@ sink { } ``` +### CDC 变更日志写入 + +当上游是变更日志类数据源（例如 MySQL CDC、PostgreSQL CDC）时，行中会包含 `INSERT`、`UPDATE_BEFORE`/`UPDATE_AFTER` +和 `DELETE` 事件。需要配置 `primary_key`，让接收端能够把 `UPDATE`、`DELETE` 行映射到正确的目标行；当目标表自身 +没有去重引擎时，配合 `support_upsert = true` 可以改为查询主键后再写入。 + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + schema = { + fields { + pk_id = bigint + name = string + score = int + } + } + rows = [ + { kind = INSERT, fields = [1, "A", 100] } + { kind = INSERT, fields = [2, "B", 100] } + { kind = INSERT, fields = [3, "C", 100] } + { kind = UPDATE_BEFORE, fields = [1, "A", 100] } + { kind = UPDATE_AFTER, fields = [1, "A_1", 100] } + { kind = DELETE, fields = [2, "B", 100] } + ] + } +} + +sink { + Clickhouse { + host = "localhost:8123" + database = "default" + table = "sink_table" + username = "default" + password = "" + primary_key = "pk_id" + } +} +``` + ### 多表写入案例 在ClickHouse中提前创建下面两张数据表： diff --git a/docs/zh/connectors/sink/Easysearch.md b/docs/zh/connectors/sink/Easysearch.md index 109217a7c1..633acf338e 100644 --- a/docs/zh/connectors/sink/Easysearch.md +++ b/docs/zh/connectors/sink/Easysearch.md @@ -15,11 +15,13 @@ import ChangeLog from '../changelog/connector-easysearch.md'; ## 使用依赖 > 依赖 > [easysearch-client](https://central.sonatype.com/artifact/com.infinilabs/easysearch-client) + ## 主要特性 - [ ] [精确一次](../../introduction/concepts/connector-v2-features.md) - [x] [变更数据捕获](../../introduction/concepts/connector-v2-features.md) - [x] [批处理](../../introduction/concepts/connector-v2-features.md) +- [ ] [支持多表写入](../../introduction/concepts/connector-v2-features.md) :::提示 @@ -45,26 +47,26 @@ import ChangeLog from '../changelog/connector-easysearch.md'; ## 接收器选项 -| 名称 | 类型 | 是否必填 | 默认值 | -|------------------------|---------|----|---------------| -| hosts | array | 是 | - | -| index | string | 是 | - | -| primary_keys | list | 否 | - | -| key_delimiter | string | 否 | `_` | -| key_encoding | enum | 否 | LEGACY | -| username | string | 否 | - | -| password | string | 否 | - | -| max_retry_count | int | 否 | 3 | -| max_batch_size | int | 否 | 10 | -| tls_verify_certificate | boolean | 否 | true | -| tls_verify_hostname | boolean | 否 | true | -| tls_keystore_path | string | 否 | - | -| tls_keystore_password | string | 否 | - | -| tls_truststore_path | string | 否 | - | -| tls_truststore_password | string | 否 | - | -| schema_save_mode | enum | 否 | CREATE_SCHEMA_WHEN_NOT_EXIST | -| data_save_mode | enum | 否 | APPEND_DATA | -| common-options | config | 否 | - | +| 名称 | 类型 | 是否必填 | 默认值 | 描述 | +|------------------------|---------|----|---------------|-------------------------------------------------------------------------------------------------------------------| +| hosts | array | 是 | - | Easysearch HTTP 集群地址，格式为 `host:port`，可指定多个主机，例如 `["host1:9200", "host2:9200"]`。 | +| index | string | 是 | - | Easysearch 索引名称。支持占位符，例如 `seatunnel_${age}`。 | +| primary_keys | list | 否 | - | 用于构建文档 `_id` 的主键字段。写入需要更新或删除语义的 CDC 数据时需要配置。 | +| key_delimiter | string | 否 | `_` | 在构建 `_id` 时连接复合主键字段的分隔符。 | +| key_encoding | enum | 否 | LEGACY | 文档主键的编码方式，详见下方 `key_encoding` 章节。 | +| username | string | 否 | - | 安全用户名。 | +| password | string | 否 | - | 安全密码。 | +| max_retry_count | int | 否 | 3 | 单次批量请求的最大重试次数。 | +| max_batch_size | int | 否 | 10 | 单次批量请求最多缓存的文档数量。 | +| tls_verify_certificate | boolean | 否 | true | 是否校验 HTTPS 证书。 | +| tls_verify_hostname | boolean | 否 | true | 是否校验 HTTPS 主机名。 | +| tls_keystore_path | string | 否 | - | PEM 或 JKS 密钥存储路径。 | +| tls_keystore_password | string | 否 | - | 密钥存储的密码。 | +| tls_truststore_path | string | 否 | - | PEM 或 JKS 信任存储路径。 | +| tls_truststore_password | string | 否 | - | 信任存储的密码。 | +| schema_save_mode | enum | 否 | CREATE_SCHEMA_WHEN_NOT_EXIST | 在启动同步任务前如何处理目标索引，详见下方的 `schema_save_mode` 章节。 | +| data_save_mode | enum | 否 | APPEND_DATA | 在启动同步任务前如何处理目标端已有数据，详见下方的 `data_save_mode` 章节。 | +| common-options | config | 否 | - | Sink 插件通用参数，详见 [Sink 通用选项](../common-options/sink-common-options.md)。 | ### hosts [array] @@ -184,6 +186,47 @@ sink { } ``` +### 按行字段写入动态索引 + +Easysearch 未实现多表 Sink 契约。以下示例只有一张输入表，每行均携带 +`target_index` 字段。索引模板读取该字段，将两行分别写入 `st_customers` +和 `st_orders`。上游表标识不会自动转换为行字段。`target_index` 是行字段； +这里不使用保留的 `${table_name}` 元数据占位符。请预先创建目标索引，或在服务器上启用 +自动创建索引。`schema_save_mode = "IGNORE"` 将索引结构管理交给服务器。 + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + row.num = 2 + schema { + fields { + id = int + name = string + target_index = string + } + } + rows = [ + { kind = INSERT, fields = [1, "alice", "customers"] }, + { kind = INSERT, fields = [2, "order", "orders"] } + ] + } +} + +sink { + Easysearch { + hosts = ["localhost:9200"] + index = "st_${target_index}" + primary_keys = ["id"] + schema_save_mode = "IGNORE" + } +} +``` + SSL (禁用证书验证) ```hocon diff --git a/docs/zh/connectors/sink/MongoDB.md b/docs/zh/connectors/sink/MongoDB.md index 7f8bee45ba..cbf4865630 100644 --- a/docs/zh/connectors/sink/MongoDB.md +++ b/docs/zh/connectors/sink/MongoDB.md @@ -316,6 +316,66 @@ sink { } ``` +### CDC 变更日志写入 + +当上游是变更日志类数据源时，可以将 `upsert-enable` 设为 `true` 并配置 `primary-key`，让 `UPDATE`、`DELETE` +事件匹配已有文档而不是作为新文档重复插入。接收端还提供 `buffer-flush.max-rows`、`buffer-flush.interval` 和 +`retry.max` 用于控制批量刷新与重试行为。 + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + schema = { + fields { + id = int + name = string + } + } + rows = [ + { kind = INSERT, fields = [1, "alice"] } + { kind = INSERT, fields = [2, "bob"] } + { kind = UPDATE_BEFORE, fields = [1, "alice"] } + { kind = UPDATE_AFTER, fields = [1, "alice_1"] } + { kind = DELETE, fields = [2, "bob"] } + ] + } +} + +sink { + MongoDB { + uri = "mongodb://user:[email protected]:27017/test_db?retryWrites=true" + database = "test_db" + collection = "users" + upsert-enable = true + primary-key = ["id"] + buffer-flush.max-rows = 2000 + buffer-flush.interval = 1000 + } +} +``` + +### 事务写入（MongoDB 4.2+） + +将 `transaction` 设为 `true` 时，批量写入会被包装成 MongoDB 多文档事务。该选项需要 MongoDB 4.2 及以上版本， +并按 checkpoint 边界提交。注意在事务模式下，引擎层的定时刷新会被自动禁用，以保证 checkpoint 提交边界不被 +打散。 + +```hocon +sink { + MongoDB { + uri = "mongodb://user:[email protected]:27017/test_db?replicaSet=rs0" + database = "test_db" + collection = "orders" + transaction = true + } +} +``` + ## 更新日志 <ChangeLog /> diff --git a/docs/zh/connectors/sink/Redis.md b/docs/zh/connectors/sink/Redis.md index e4a97f7668..64c5bd7edb 100644 --- a/docs/zh/connectors/sink/Redis.md +++ b/docs/zh/connectors/sink/Redis.md @@ -217,6 +217,24 @@ sink { } ``` +### 按上游字段动态生成键 + +将 `support_custom_key` 设为 `true` 后，`key` 中可以包含占位符 `${field}`（同时也兼容旧的 `{field}` 写法）， +连接器在写入时会把占位符替换为对应上游字段的值。常用于把同一 Redis 数据类型按上游主键写入不同键的场景。 + +```hocon +sink { + Redis { + host = "localhost" + port = 6379 + key = "person:${id}" + support_custom_key = true + data_type = KEY + format = JSON + } +} +``` + ## 变更日志 <ChangeLog /> diff --git a/docs/zh/connectors/source/Clickhouse.md b/docs/zh/connectors/source/Clickhouse.md index 2625a97ed2..a061a4841c 100644 --- a/docs/zh/connectors/source/Clickhouse.md +++ b/docs/zh/connectors/source/Clickhouse.md @@ -195,26 +195,30 @@ sink { ``` ### 多表配置 + +当 `table_list` 包含多个条目时，SeaTunnel 会为每个上游表生成一个独立的数据流，下游连接器可以使用 `${table_name}` +占位符将每条数据流路由到对应的目标表。结合 Clickhouse 接收端的 `table = "${table_name}_multi_table_sink"` 等 +配置，可以把每个上游表分别写入各自的目标表。 + ```hocon env { job.mode = "BATCH" - parallelism = 5 + parallelism = 3 } source { Clickhouse { host = "localhost:8123" - username = "xxx" - password = "xxx" + username = "default" + password = "" table_list = [ { - table_path = "default.table1" - sql = "select * from default.table1 where id > 2 and type = 1" + table_path = "default.source_table" + sql = "select * from source_table" }, { - table_path = "default.table2" - sql = "select * from default.table2 where age > 18" - split_size = 1 + table_path = "default.source_merge_tree_table" + filter_query = "id < 47" } ] server_time_zone = "UTC" @@ -224,14 +228,42 @@ source { } } -# Console printing of the read Clickhouse data +# 每个上游表写入一个独立的目标表 sink { - Console { - parallelism = 1 + Clickhouse { + host = "localhost:8123" + database = "default" + table = "${table_name}_multi_table_sink" + username = "default" + password = "" } } ``` +## 常见问题 + +### 如何只读取 ClickHouse 表的部分分区或部分行？ + +优先使用 `partition_list` 按分区过滤，或使用 `filter_query` 在服务端下推谓词；这两种方式都能在大表上避免全表扫描。 +当同时需要投影特定列时，可以把它们和 `sql` 一起配置；此时连接器会进入 SQL 模式，并结合分区/过滤提示一起执行用户查询。 + +### 什么时候使用 SQL 模式，什么时候使用查表模式？ + +需要 join、聚合、子查询或多表投影时使用 SQL 模式（设置了 `sql`）。只需按主键读取单张逻辑表并希望基于 +`system.parts` 做并行切分时使用查表模式（仅设置 `table_path`）。对于必须在单分片上运行的 SQL，连接器会自动 +降级为单并发，即使配置了更高的 `parallelism` 也不会真正并行。 + +### 如何调节吞吐和内存压力？ + +两个主要参数是 `split.size`（每个 SeaTunnel split 合并多少 ClickHouse part）和 `batch_size`（每次往返拉取 +的行数）。在小 part 较多的场景调小 `split.size` 可以提高并行度；在 part 较大的场景调大 `batch_size` 可以摊薄 +往返开销。`clickhouse.config` 这个 map 暴露了所有 `clickhouse-jdbc` 参数（例如 `socket_timeout`）供进一步调优。 + +### `table_list` 和单个 `table_path` 是什么关系？ + +当配置只对应一张逻辑表时，可以把表级配置直接展平到外层，并把参数写成 `split.size`（注意带点）。当 `table_list` +中有多张表时，每张表的选项都要写在各自的条目里，并在条目内继续使用 `split_size`。 + ## 变更日志 <ChangeLog /> diff --git a/docs/zh/connectors/source/Easysearch.md b/docs/zh/connectors/source/Easysearch.md index c5841d4400..45250c5d31 100644 --- a/docs/zh/connectors/source/Easysearch.md +++ b/docs/zh/connectors/source/Easysearch.md @@ -26,6 +26,7 @@ import ChangeLog from '../changelog/connector-easysearch.md'; - [x] [列映射](../../introduction/concepts/connector-v2-features.md) - [ ] [并行度](../../introduction/concepts/connector-v2-features.md) - [ ] [支持用户自定义拆分](../../introduction/concepts/connector-v2-features.md) +- [ ] [支持多表读](../../introduction/concepts/connector-v2-features.md) :::提示 diff --git a/docs/zh/connectors/source/Http.md b/docs/zh/connectors/source/Http.md index 588e710e20..03adc5767a 100644 --- a/docs/zh/connectors/source/Http.md +++ b/docs/zh/connectors/source/Http.md @@ -721,6 +721,73 @@ source { ``` +## 流处理模式 + +在 `STREAMING` 任务中，连接器会以固定间隔轮询配置的 URL，使 Source 表现为持续运行的拉取任务。 +`poll_interval_millis` 控制两次请求之间的间隔；`retry`、`retry_backoff_multiplier_ms`、`retry_backoff_max_ms` +共同决定遇到 `IOException` 时的重试行为。 + +```hocon +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 60000 +} + +source { + Http { + url = "https://api.example.com/events"; + method = "GET" + format = "json" + poll_interval_millis = 5000 + retry = 3 + retry_backoff_multiplier_ms = 200 + retry_backoff_max_ms = 5000 + schema = { + fields { + id = string + type = string + timestamp = string + } + } + } +} + +sink { + Console {} +} +``` + +## 常见问题 + +### 如何通过 HTTP 下载二进制文件（PDF、图片、ZIP）？ + +将 `format` 设置为 `binary`。响应体会按 `binary_chunk_size` 字节（默认 10 MiB）切分，并以 +`(data: bytes, relativePath: string, partIndex: long)` 三列的形式输出。在 BATCH 模式下配合 `LocalFile` +接收端即可写入磁盘。 + +### 如何让分页字段作为 URL 参数传递？ + +将 `keep_page_param_as_http_param` 设置为 `true`，连接器会把分页字段写入 `params`，作为查询串参数出现在每次 +请求中。设置为 `false` 时，连接器只会更新 `headers`、`params`、`body` 中已存在的键或 `${page}` 占位符。 + +### `use_placeholder_replacement` 的实际作用是什么？ + +为 `true` 时，连接器会在 headers、params 和 body 中识别 `${page}` 和 `${cursor}` 占位符并替换为真实值， +也支持带前缀/后缀的形式（如 `"10${page}"` 在 `page=5` 时变为 `"105"`）。为 `false` 时，仅按键名替换： +分页字段必须已经作为键存在于目标 map 中。 + +### 为什么 `method = POST` 且没有设置 `body` 时请求体看起来是空的？ + +`keep_params_as_form = false` 时默认走非表单分支，会将 `body` 序列化为 JSON。当未配置 `body` 时，连接器仍 +会发送 JSON 对象 `{}` 作为请求体，因此 HTTP 服务端能收到合法的 JSON 负载。如需切换到表单分支，可将 +`keep_params_as_form` 设为 `true`（并按需设置 `Content-Type: application/x-www-form-urlencoded`）。 + +### 如何限制重试并放慢重连节奏？ + +`retry` 用于设置 `IOException` 时的最大重试次数；`retry_backoff_multiplier_ms`（默认 `100`）是退避基准 +（毫秒）；`retry_backoff_max_ms`（默认 `10000`）用于限制最大退避，避免误配置导致任务长时间等待。 + ## 变更日志 <ChangeLog /> diff --git a/docs/zh/connectors/source/MongoDB.md b/docs/zh/connectors/source/MongoDB.md index 7d7f0d9a86..eda46e3fc8 100644 --- a/docs/zh/connectors/source/MongoDB.md +++ b/docs/zh/connectors/source/MongoDB.md @@ -373,6 +373,54 @@ sink { } ``` +### 基于分区键的并行读取 + +要在大型集合上加速扫描，可以设置分区键 `partition.split-key` 和目标分片大小 `partition.split-size`， +连接器会按分区键顺序读取文档，并将每个分片交给并行的读取任务。`fetch.size` 控制每次往返服务端返回的文档数， +需要和分区大小一起调优。 + +```hocon +source { + MongoDB { + uri = "mongodb://user:[email protected]:27017" + database = "test_db" + collection = "orders" + partition.split-key = "_id" + partition.split-size = 1024 + fetch.size = 2048 + max.time-min = 30 + schema = { + fields { + _id = string + status = string + amount = double + } + } + } +} +``` + +### 从副本集或分片集群读取 + +连接器接受任意标准 MongoDB URI。副本集场景使用 `replicaSet=xxx`；如果凭据存储在非默认的 auth 数据库， +追加 `authSource=admin`。 + +```hocon +source { + MongoDB { + uri = "mongodb://user:password@mongo1:27017,mongo2:27017,mongo3:27017/test_db?replicaSet=rs0&authSource=admin&readPreference=secondary" + database = "test_db" + collection = "orders" + schema = { + fields { + _id = string + status = string + } + } + } +} +``` + ## 修改日志 <ChangeLog /> diff --git a/docs/zh/connectors/source/Redis.md b/docs/zh/connectors/source/Redis.md index ff9064280b..6b9768b4e2 100644 --- a/docs/zh/connectors/source/Redis.md +++ b/docs/zh/connectors/source/Redis.md @@ -560,6 +560,34 @@ sink { } ``` +### 使用自定义键名和值列名读取 + +通过 `key_field_name`、`single_field_name` 和 `read_key_enabled = true` 可以把 Redis 键纳入输出，并控制其 +值映射到 schema 中的哪一列。对于基础值类型（`string`、`list`、`set`、`zset`），必须设置 `single_field_name`， +让连接器知道哪个 schema 列接收该值。 + +```hocon +source { + Redis { + host = "localhost" + port = 6379 + keys = "string_test*" + data_type = string + read_key_enabled = true + key_field_name = custom_key + single_field_name = custom_value + format = json + schema = { + table = "RedisDatabase.RedisTable" + columns = [ + { name = "custom_key", type = "string" } + { name = "custom_value", type = "string" } + ] + } + } +} +``` + ## 变更日志 <ChangeLog />
