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 740494d9a9 [Docs][Connector-V2] Improve Easysearch IoTDB and Hudi docs
(#11294)
740494d9a9 is described below
commit 740494d9a9a7ea2a1cd49f1c41270b60465516b9
Author: Daniel Carter <[email protected]>
AuthorDate: Fri Jul 3 18:14:11 2026 +0800
[Docs][Connector-V2] Improve Easysearch IoTDB and Hudi docs (#11294)
Co-authored-by: DanielCarter-stack
<[email protected]>
---
docs/en/connectors/sink/Easysearch.md | 61 ++++++++-------
docs/en/connectors/sink/Hudi.md | 78 +++++++++++++++----
docs/en/connectors/sink/IoTDB.md | 4 +-
docs/en/connectors/source/Easysearch.md | 88 ++++++++++++++-------
docs/en/connectors/source/IoTDB.md | 26 ++++---
docs/zh/connectors/sink/Easysearch.md | 95 ++++++++++++-----------
docs/zh/connectors/sink/Hudi.md | 130 +++++++++++++++++++++-----------
docs/zh/connectors/sink/IoTDB.md | 24 +++---
docs/zh/connectors/source/Easysearch.md | 107 ++++++++++++++++----------
docs/zh/connectors/source/IoTDB.md | 7 +-
10 files changed, 399 insertions(+), 221 deletions(-)
diff --git a/docs/en/connectors/sink/Easysearch.md
b/docs/en/connectors/sink/Easysearch.md
index 2f12958106..242995b280 100644
--- a/docs/en/connectors/sink/Easysearch.md
+++ b/docs/en/connectors/sink/Easysearch.md
@@ -10,16 +10,17 @@ import ChangeLog from
'../changelog/connector-easysearch.md';
## Description
-A sink plugin which use send data to `INFINI Easysearch`.
+A sink plugin used to send data to `INFINI Easysearch`.
## Using Dependency
-> Depenndency
[easysearch-client](https://central.sonatype.com/artifact/com.infinilabs/easysearch-client)
->
- ## Key features
+> Dependency
[easysearch-client](https://central.sonatype.com/artifact/com.infinilabs/easysearch-client)
+
+## Key features
- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
- [x] [cdc](../../introduction/concepts/connector-v2-features.md)
+- [x] [batch](../../introduction/concepts/connector-v2-features.md)
:::tip
@@ -41,7 +42,7 @@ Engine Supported
| LONG | LONG |
| FLOAT<br/>HALF_FLOAT | FLOAT |
| DOUBLE | DOUBLE |
-| Date | LOCAL_DATE_TIME_TYPE |
+| DATE | LOCAL_DATE_TIME_TYPE |
## Sink Options
@@ -71,12 +72,11 @@ Engine Supported
### index [string]
-`INFINI Easysearch` `index` name.Index support contains variables of field
name,such as `seatunnel_${age}`,and the field must appear at seatunnel row.
-If not, we will treat it as a normal index.
+`INFINI Easysearch` index name. The index can contain field placeholders, such
as `seatunnel_${age}`. The referenced field must exist in the input row;
otherwise the value is treated as a normal literal index name.
### primary_keys [list]
-Primary key fields used to generate the document `_id`, this is cdc required
options.
+Primary key fields used to generate the document `_id`. Configure this option
when writing CDC rows that need update or delete semantics.
### key_delimiter [string]
@@ -92,11 +92,11 @@ security password
### max_retry_count [int]
-one bulk request max try size
+Maximum retry count for one bulk request.
### max_batch_size [int]
-batch bulk doc max size
+Maximum number of documents buffered in one bulk request.
### tls_verify_certificate [boolean]
@@ -143,28 +143,38 @@ Sink plugin common parameters, please refer to [Sink
Common Options](../common-o
## Examples
-Simple
+### Write To One Index
-```bash
+```hocon
sink {
- Easysearch {
- hosts = ["localhost:9200"]
- index = "seatunnel-${age}"
- }
+ Easysearch {
+ hosts = ["localhost:9200"]
+ index = "seatunnel_index"
+ max_batch_size = 100
+ }
}
```
-CDC(Change data capture) event
+### Write To Dynamic Index
-```bash
+```hocon
sink {
- Easysearch {
- hosts = ["localhost:9200"]
- index = "seatunnel-${age}"
+ Easysearch {
+ hosts = ["localhost:9200"]
+ index = "seatunnel_${age}"
+ }
+}
+```
- # cdc required options
- primary_keys = ["key1", "key2", ...]
- }
+### CDC Event
+
+```hocon
+sink {
+ Easysearch {
+ hosts = ["localhost:9200"]
+ index = "seatunnel_${age}"
+ primary_keys = ["key1", "key2"]
+ }
}
```
@@ -211,7 +221,7 @@ sink {
}
```
-SAVE_MODE
+### Save Mode
```hocon
sink {
@@ -229,4 +239,3 @@ sink {
## Changelog
<ChangeLog />
-
diff --git a/docs/en/connectors/sink/Hudi.md b/docs/en/connectors/sink/Hudi.md
index 8f604fb07c..1e7827c1b7 100644
--- a/docs/en/connectors/sink/Hudi.md
+++ b/docs/en/connectors/sink/Hudi.md
@@ -13,6 +13,7 @@ Used to write data to Hudi.
- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
- [x] [cdc](../../introduction/concepts/connector-v2-features.md)
- [x] [support multiple table
write](../../introduction/concepts/connector-v2-features.md)
+- [x] [timer flush](../../introduction/concepts/connector-v2-features.md)
## Options
@@ -24,6 +25,7 @@ Base configuration:
| conf_files_path | string | no | -
|
| table_list | Array | no | -
|
| schema_save_mode | enum | no |
CREATE_SCHEMA_WHEN_NOT_EXIST|
+| data_save_mode | enum | no | APPEND_DATA
|
| common-options | Config | no | -
|
Table list configuration:
@@ -80,7 +82,7 @@ Note: When this configuration corresponds to a single table,
you can flatten the
### index_type [string]
-`index_type` The index type of hudi table. Currently, `BLOOM`, `SIMPLE`, and
`GLOBAL SIMPLE` are supported.
+`index_type` The index type of hudi table. Currently, `BLOOM`, `SIMPLE`, and
`GLOBAL_BLOOM` are supported.
### index_class_name [string]
@@ -100,11 +102,11 @@ Note: When this configuration corresponds to a single
table, you can flatten the
### batch_interval_ms [Int]
-`batch_interval_ms` The interval time of batch write to hudi table.
+`batch_interval_ms` The maximum interval, in milliseconds, between two flushes
to Hudi.
### batch_size [Int]
-`batch_size` The size of batch write to hudi table.
+`batch_size` The maximum number of rows buffered before one flush to Hudi.
### insert_shuffle_parallelism [Int]
@@ -135,30 +137,58 @@ Option introduction：
`ERROR_WHEN_SCHEMA_NOT_EXIST` ：Error will be reported when the table does not
exist
`IGNORE` ：Ignore the treatment of the table
+### data_save_mode [Enum]
+
+Choose how to handle existing data before the synchronization task starts.
+
+`DROP_DATA`: Keep the table structure and delete existing data.
+
+`APPEND_DATA`: Keep the table structure and existing data.
+
+`ERROR_WHEN_DATA_EXISTS`: Throw an error when data already exists.
+
### common options
Sink plugin common parameters, please refer to [Sink Common
Options](../common-options/sink-common-options.md) for details.
## Examples
-### single table
+### Single Table Upsert
+
+When `op_type` is `UPSERT`, `record_key_fields` must be configured.
+
```hocon
sink {
Hudi {
- table_dfs_path = "hdfs://nameserivce/data/"
+ table_dfs_path = "/tmp/seatunnel_mnt/hudi"
database = "st"
- table_name = "test_table"
+ table_name = "st_test"
table_type = "COPY_ON_WRITE"
- conf_files_path =
"/home/test/hdfs-site.xml;/home/test/core-site.xml;/home/test/yarn-site.xml"
- batch_size = 10000
- use.kerberos = true
- kerberos.principal = "test_user@xxx"
- kerberos.principal.file = "/home/test/test_user.keytab"
+ op_type = "UPSERT"
+ record_key_fields = "c_bigint"
+ batch_size = 1000
+ batch_interval_ms = 1000
}
}
```
-### Multiple table
+### Minimal Single Table
+
+For append-only writes, only `table_dfs_path` and `table_name` are required.
+
+```hocon
+sink {
+ Hudi {
+ table_dfs_path = "/tmp/seatunnel_mnt/hudi"
+ table_name = "st_test"
+ }
+}
+```
+
+### Multiple Tables
+
+Use `table_list` when the upstream source produces multiple tables.
+
```hocon
env {
parallelism = 1
@@ -188,15 +218,14 @@ sink {
database = "st1"
table_name = "role"
table_type = "COPY_ON_WRITE"
- op_type="INSERT"
+ op_type = "INSERT"
batch_size = 10000
},
{
database = "st1"
table_name = "user"
table_type = "COPY_ON_WRITE"
- op_type="UPSERT"
- # op_type is 'UPSERT', must configured record_key_fields
+ op_type = "UPSERT"
record_key_fields = "user_id"
batch_size = 10000
},
@@ -206,7 +235,24 @@ sink {
table_type = "MERGE_ON_READ"
}
]
- ...
+ }
+}
+```
+
+### CDC To Hudi
+
+Enable `cdc_enabled` when the target Hudi table needs to persist CDC changelog
information.
+
+```hocon
+sink {
+ Hudi {
+ table_dfs_path = "/tmp/seatunnel_mnt/hudi"
+ database = "st"
+ table_name = "st_test"
+ table_type = "COPY_ON_WRITE"
+ op_type = "UPSERT"
+ record_key_fields = "id"
+ cdc_enabled = true
}
}
```
diff --git a/docs/en/connectors/sink/IoTDB.md b/docs/en/connectors/sink/IoTDB.md
index b5f5efc8d5..d8728d2174 100644
--- a/docs/en/connectors/sink/IoTDB.md
+++ b/docs/en/connectors/sink/IoTDB.md
@@ -49,9 +49,9 @@ Used to write data to IoTDB.
| password | String | Yes | -
| IoTDB user password
|
| key_device | String | Yes | -
| Specify field name of the IoTDB deviceId in SeaTunnelRow
|
| key_timestamp | String | No | processing time
| Specify field-name of the IoTDB timestamp in SeaTunnelRow. If not
specified, use processing-time as timestamp
|
-| key_measurement_fields | Array | No | exclude `device` &
`timestamp` | Specify field-name of the IoTDB measurement list in SeaTunnelRow.
If not specified, include all fields but exclude `device` & `timestamp`
|
+| key_measurement_fields | Array | No | exclude device and
timestamp fields | Specify field names of the IoTDB measurement list in
SeaTunnelRow. If not specified, include all fields except `key_device` and
`key_timestamp` fields. |
| storage_group | Array | No | -
| Specify device storage group(path prefix) <br/> example: deviceId =
\${storage_group} + "." + \${key_device}
|
-| batch_size | Integer | No | 1024
| For batch writing, when the number of buffers reaches the number of
`batch_size` or the time reaches `batch_interval_ms`, the data will be flushed
into the IoTDB |
+| batch_size | Integer | No | 1024
| For batch writing, data is flushed into IoTDB when the buffered row
count reaches `batch_size`.
|
| max_retries | Integer | No | -
| The number of retries to flush failed
|
| retry_backoff_multiplier_ms | Integer | No | -
| Using as a multiplier for generating the next delay for backoff
|
| max_retry_backoff_ms | Integer | No | -
| The amount of time to wait before attempting to retry a request to
`IoTDB`
|
diff --git a/docs/en/connectors/source/Easysearch.md
b/docs/en/connectors/source/Easysearch.md
index b27bdcf150..2105cc284d 100644
--- a/docs/en/connectors/source/Easysearch.md
+++ b/docs/en/connectors/source/Easysearch.md
@@ -16,7 +16,7 @@ Used to read data from INFINI Easysearch.
## Using Dependency
-> Depenndency
[easysearch-client](https://central.sonatype.com/artifact/com.infinilabs/easysearch-client)
+> Dependency
[easysearch-client](https://central.sonatype.com/artifact/com.infinilabs/easysearch-client)
## Key features
@@ -47,7 +47,28 @@ Engine Supported
| LONG | LONG |
| FLOAT<br/>HALF_FLOAT | FLOAT |
| DOUBLE | DOUBLE |
-| Date | LOCAL_DATE_TIME_TYPE |
+| DATE | LOCAL_DATE_TIME_TYPE |
+
+## Source Options
+
+| name | type | required | default value | description |
+|------------------------|---------|----------|---------------|-------------|
+| hosts | array | yes | - | Easysearch
HTTP addresses. |
+| index | string | yes | - | Easysearch
index name. Wildcard matching such as `seatunnel-*` is supported. |
+| username | string | no | - | Username for
secured Easysearch clusters. |
+| password | string | no | - | Password for
secured Easysearch clusters. |
+| source | array | no | - | Fields to read
from the index. Configure either `source` or `schema`. |
+| schema | config | no | - | SeaTunnel
schema used to read and convert fields. Configure either `schema` or `source`. |
+| query | json | no | `{"match_all":{}}` |
Easysearch DSL query used to filter records. |
+| scroll_time | string | no | 1m | Time that
Easysearch keeps the scroll context alive. |
+| scroll_size | int | no | 100 | Maximum
records returned by each scroll 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. |
+| common-options | config | no | - | Source plugin
common options. |
### hosts [array]
@@ -69,7 +90,7 @@ Easysearch index name, support * fuzzy matching.
The fields of index.
You can get the document id by specifying the field `_id`.If sink _id to other
index,you need specify an alias for _id due to the Easysearch limit.
-If you don't config source, you must config `schema`.
+If `source` is not configured, `schema` must be configured.
### query [json]
@@ -86,8 +107,8 @@ Maximum number of hits to be returned with each Easysearch
scroll request.
### schema
-The structure of the data, including field names and field types.
-If you don't config schema, you must config `source`.
+The structure of the data, including field names and field types. For more
details, see [Schema Feature](../../introduction/concepts/schema-feature.md).
+If `schema` is not configured, `source` must be configured.
### tls_verify_certificate [boolean]
@@ -119,42 +140,51 @@ Source plugin common parameters, please refer to [Source
Common Options](../comm
## Examples
-simple
+### Read Selected Fields
```hocon
-Easysearch {
+source {
+ Easysearch {
hosts = ["localhost:9200"]
index = "seatunnel-*"
- source = ["_id","name","age"]
- query = {"range":{"firstPacket":{"gte":1700407367588,"lte":1700407367588}}}
+ source = ["_id", "name", "age"]
+ query = {"range": {"age": {"gte": 18, "lte": 60}}}
+ }
}
```
-complex
+### Read With Schema And Query
```hocon
-Easysearch {
- hosts = ["Easysearch:9200"]
+source {
+ Easysearch {
+ hosts = ["https://e2e_easysearch:9200";]
+ username = "admin"
+ password = "admin"
+ tls_verify_certificate = false
+ tls_verify_hostname = false
+
index = "st_index"
+ query = {"range": {"c_int": {"gte": 10, "lte": 20}}}
schema = {
- fields {
- c_map = "map<string, tinyint>"
- c_array = "array<tinyint>"
- c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
- c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(2, 1)"
- c_bytes = bytes
- c_date = date
- c_timestamp = timestamp
- }
+ fields {
+ c_map = "map<string, tinyint>"
+ c_array = "array<tinyint>"
+ c_string = string
+ c_boolean = boolean
+ c_tinyint = tinyint
+ c_smallint = smallint
+ c_int = int
+ c_bigint = bigint
+ c_float = float
+ c_double = double
+ c_decimal = "decimal(2, 1)"
+ c_bytes = bytes
+ c_date = date
+ c_timestamp = timestamp
+ }
}
- query = {"range":{"firstPacket":{"gte":1700407367588,"lte":1700407367588}}}
+ }
}
```
diff --git a/docs/en/connectors/source/IoTDB.md
b/docs/en/connectors/source/IoTDB.md
index a95328e562..1ab8aa3e98 100644
--- a/docs/en/connectors/source/IoTDB.md
+++ b/docs/en/connectors/source/IoTDB.md
@@ -42,6 +42,10 @@ Used to read data from IoTDB.
| FLOAT | FLOAT |
| DOUBLE | DOUBLE |
| TEXT | STRING |
+| STRING | STRING |
+| TIMESTAMP | TIMESTAMP |
+| BLOB | STRING |
+| DATE | DATE |
## Source Options
@@ -52,14 +56,14 @@ Used to read data from IoTDB.
| password | string | yes | - | IoTDB user
password
|
| sql | string | yes | - | execute
sql statement
|
| schema | config | yes | - | The data
schema. For more details, please refer to [Schema
Feature](../../introduction/concepts/schema-feature.md).
|
-| fetch_size | int | no | - | the
fetch_size of the IoTDB when you select
|
-| lower_bound | long | no | - | the
lower_bound of the IoTDB when you select
|
-| upper_bound | long | no | - | the
upper_bound of the IoTDB when you select
|
-| num_partitions | int | no | - | the
num_partitions of the IoTDB when you select
|
-| thrift_default_buffer_size | int | no | - | the
thrift_default_buffer_size of the IoTDB when you select
|
-| thrift_max_frame_size | int | no | - | the thrift
max frame size
|
-| enable_cache_leader | boolean | no | - |
enable_cache_leader of the IoTDB when you select
|
-| version | string | no | - | SQL
semantic version used by the client, The possible values are: `V_0_12`,
`V_0_13` |
+| fetch_size | int | no | - | Number of
rows fetched from IoTDB in one request.
|
+| lower_bound | long | no | - | Lower time
bound used when SeaTunnel splits the query by time.
|
+| upper_bound | long | no | - | Upper time
bound used when SeaTunnel splits the query by time.
|
+| num_partitions | int | no | - | Number of
time-range partitions. Use it together with `lower_bound` and `upper_bound`.
|
+| thrift_default_buffer_size | int | no | - | Initial
Thrift buffer size for the IoTDB client.
|
+| thrift_max_frame_size | int | no | - | Maximum
Thrift frame size for the IoTDB client.
|
+| enable_cache_leader | boolean | no | - | Whether to
cache the leader node in the IoTDB client.
|
+| version | string | no | - | SQL
semantic version used by the client. The possible values are `V_0_12` and
`V_0_13`. |
| common-options | | no | - | Source
plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details |
We can use time column as a partition key in SQL queries.
@@ -103,6 +107,9 @@ source {
username = "root"
password = "root"
sql = "SELECT temperature, moisture, c_int, c_bigint, c_float, c_double,
c_string, c_boolean FROM root.test_group.* WHERE time < 4102329600000 align by
device"
+ lower_bound = 1
+ upper_bound = 4102329600000
+ num_partitions = 10
schema {
fields {
ts = timestamp
@@ -126,6 +133,8 @@ sink {
}
```
+`lower_bound`, `upper_bound`, and `num_partitions` are optional. They are
useful when the query covers a large time range and you want SeaTunnel to split
the read into multiple time partitions.
+
The data format from upstream IoTDB is as follows:
```shell
@@ -150,4 +159,3 @@ The data format loaded to SeaTunnelRow is as follows:
## Changelog
<ChangeLog />
-
diff --git a/docs/zh/connectors/sink/Easysearch.md
b/docs/zh/connectors/sink/Easysearch.md
index e33249a65c..2cd73ce51f 100644
--- a/docs/zh/connectors/sink/Easysearch.md
+++ b/docs/zh/connectors/sink/Easysearch.md
@@ -10,22 +10,22 @@ import ChangeLog from
'../changelog/connector-easysearch.md';
## 描述
-一个使用将数据发送到 `INFINI Easysearch` 的接收器插件.
+用于将数据写入 `INFINI Easysearch`。
## 使用依赖
> 依赖
> [easysearch-client](https://central.sonatype.com/artifact/com.infinilabs/easysearch-client)
->
- ## 关键特性
+## 主要特性
- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md)
-- [x] [cdc](../../introduction/concepts/connector-v2-features.md)
+- [x] [变更数据捕获](../../introduction/concepts/connector-v2-features.md)
+- [x] [批处理](../../introduction/concepts/connector-v2-features.md)
:::提示
支持的引擎
-* 支持 [INFINI Easysearch](https://www.infini.com/download/?product=easysearch)
发布的所有版本.
+* 支持 [INFINI Easysearch](https://www.infini.com/download/?product=easysearch)
发布的所有版本。
:::
@@ -41,42 +41,41 @@ import ChangeLog from
'../changelog/connector-easysearch.md';
| LONG | LONG |
| FLOAT<br/>HALF_FLOAT | FLOAT |
| DOUBLE | DOUBLE |
-| Date | LOCAL_DATE_TIME_TYPE |
+| DATE | LOCAL_DATE_TIME_TYPE |
## 接收器选项
-| 名称 | 类型 | 必需 | 默认值 |
+| 名称 | 类型 | 是否必填 | 默认值 |
|------------------------|---------|----|---------------|
| hosts | array | 是 | - |
| index | string | 是 | - |
-| primary_keys | list | 否 | |
-| key_delimiter | string | 否 | `_` |
-| 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 | | 否 | - |
+| primary_keys | list | 否 | - |
+| key_delimiter | string | 否 | `_` |
+| 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]
-`INFINI Easysearch` 集群http地址，格式为 `host:port` , 允许指定多个主机.例如 `["host1:9200",
"host2:9200"]`.
+`INFINI Easysearch` 集群 HTTP 地址，格式为 `host:port`，允许指定多个主机，例如 `["host1:9200",
"host2:9200"]`。
### index [string]
-`INFINI Easysearch` `index` 名称.索引支持包含字段名变量,例如
`seatunnel_${age}`,该字段必须出现在seatunnel行.
-如果没有，我们将把它当作一个正常的索引.
+`INFINI Easysearch` 索引名称。索引名可以包含字段占位符，例如
`seatunnel_${age}`。引用的字段必须存在于输入行中；如果不存在，会按普通索引名处理。
### primary_keys [list]
-用于生成文档 `_id`的主键字段，这是cdc必需的选项.
+用于生成文档 `_id` 的主键字段。写入需要更新或删除语义的 CDC 数据时需要配置。
### key_delimiter [string]
@@ -92,11 +91,11 @@ import ChangeLog from
'../changelog/connector-easysearch.md';
### max_retry_count [int]
-一个批量请求的最大尝试大小
+单次批量请求的最大重试次数。
### max_batch_size [int]
-批量文档最大大小
+单次批量请求最多缓存的文档数量。
### tls_verify_certificate [boolean]
@@ -139,32 +138,42 @@ PEM或JKS信任存储的路径。运行SeaTunnel的操作系统用户必须能
### common options
-接收器插件常用参数，详见 [Sink Common Options](../common-options/sink-common-options.md)
+Sink 插件通用参数，详见 [Sink Common Options](../common-options/sink-common-options.md)。
## 示例
-简单的例子
+### 写入固定索引
-```bash
+```hocon
sink {
- Easysearch {
- hosts = ["localhost:9200"]
- index = "seatunnel-${age}"
- }
+ Easysearch {
+ hosts = ["localhost:9200"]
+ index = "seatunnel_index"
+ max_batch_size = 100
+ }
}
```
-CDC(变更数据捕获) 事件
+### 写入动态索引
-```bash
+```hocon
sink {
- Easysearch {
- hosts = ["localhost:9200"]
- index = "seatunnel-${age}"
+ Easysearch {
+ hosts = ["localhost:9200"]
+ index = "seatunnel_${age}"
+ }
+}
+```
- # cdc required options
- primary_keys = ["key1", "key2", ...]
- }
+### CDC 事件
+
+```hocon
+sink {
+ Easysearch {
+ hosts = ["localhost:9200"]
+ index = "seatunnel_${age}"
+ primary_keys = ["key1", "key2"]
+ }
}
```
diff --git a/docs/zh/connectors/sink/Hudi.md b/docs/zh/connectors/sink/Hudi.md
index 75024388f4..7ca61be589 100644
--- a/docs/zh/connectors/sink/Hudi.md
+++ b/docs/zh/connectors/sink/Hudi.md
@@ -8,47 +8,49 @@ import ChangeLog from '../changelog/connector-hudi.md';
用于将数据写入 Hudi。
-## 主要特点
+## 主要特性
-- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
-- [x] [cdc](../../introduction/concepts/connector-v2-features.md)
-- [x] [support multiple table
write](../../introduction/concepts/connector-v2-features.md)
+- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md)
+- [x] [变更数据捕获](../../introduction/concepts/connector-v2-features.md)
+- [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md)
+- [x] [定时刷新](../../introduction/concepts/connector-v2-features.md)
## 选项
基础配置:
-| 名称 | 名称 | 是否必需 | 默认值 |
-|----------------------------|--------|------
|------------------------------|
+| 名称 | 类型 | 是否必填 | 默认值 |
+|----------------------------|--------|----------|------------------------------|
| table_dfs_path | string | 是 | - |
| conf_files_path | string | 否 | - |
-| table_list | string | 否 | - |
+| table_list | array | 否 | - |
| schema_save_mode | enum | 否 | CREATE_SCHEMA_WHEN_NOT_EXIST |
+| data_save_mode | enum | 否 | APPEND_DATA |
| common-options | config | 否 | - |
表清单配置:
-| 名称 | 类型 | 是否必需 | 默认值 |
+| 名称 | 类型 | 是否必填 | 默认值 |
|----------------------------|--------|----------|---------------|
-| table_name | string | yes | - |
-| database | string | no | default |
-| table_type | enum | no | COPY_ON_WRITE |
-| op_type | enum | no | insert |
-| record_key_fields | string | no | - |
-| partition_fields | string | no | - |
-| precombine_field | string | no | - |
-| batch_interval_ms | Int | no | 1000 |
-| batch_size | Int | no | 1000 |
-| insert_shuffle_parallelism | Int | no | 2 |
-| upsert_shuffle_parallelism | Int | no | 2 |
-| min_commits_to_keep | Int | no | 20 |
-| max_commits_to_keep | Int | no | 30 |
-| index_type | enum | no | BLOOM |
-| index_class_name | string | no | - |
-| record_byte_size | Int | no | 1024 |
-| cdc_enabled | boolean| no | false |
-
-注意: 当此配置对应于单个表时，您可以将table_list中的配置项展平到外层。
+| table_name | string | 是 | - |
+| database | string | 否 | default |
+| table_type | enum | 否 | COPY_ON_WRITE |
+| op_type | enum | 否 | INSERT |
+| record_key_fields | string | 否 | - |
+| partition_fields | string | 否 | - |
+| precombine_field | string | 否 | - |
+| batch_interval_ms | int | 否 | 1000 |
+| batch_size | int | 否 | 1000 |
+| insert_shuffle_parallelism | int | 否 | 2 |
+| upsert_shuffle_parallelism | int | 否 | 2 |
+| min_commits_to_keep | int | 否 | 20 |
+| max_commits_to_keep | int | 否 | 30 |
+| index_type | enum | 否 | BLOOM |
+| index_class_name | string | 否 | - |
+| record_byte_size | int | 否 | 1024 |
+| cdc_enabled | boolean| 否 | false |
+
+注意：写入单表时，可以把 `table_list` 中的表配置项平铺到外层。
### table_name [string]
@@ -56,7 +58,7 @@ import ChangeLog from '../changelog/connector-hudi.md';
### database [string]
-`database` Hudi 表的database.
+`database` Hudi 表所属的数据库。
### table_dfs_path [string]
@@ -64,11 +66,11 @@ import ChangeLog from '../changelog/connector-hudi.md';
### table_type [enum]
-`table_type` Hudi 表的类型。
+`table_type` Hudi 表的类型，可选值为 `COPY_ON_WRITE` 和 `MERGE_ON_READ`。
### record_key_fields [string]
-`record_key_fields` Hudi 表的记录键字段, 当op_type是`UPSERT`类型时, 必须配置该项.
+`record_key_fields` Hudi 表的记录键字段。当 `op_type` 为 `UPSERT` 时，必须配置该项。
### partition_fields [string]
@@ -80,7 +82,7 @@ import ChangeLog from '../changelog/connector-hudi.md';
### index_type [string]
-`index_type` Hudi 表的索引类型. 当前只支持`BLOOM`, `SIMPLE`, `GLOBAL SIMPLE`三种类型.
+`index_type` Hudi 表的索引类型。当前支持 `BLOOM`、`SIMPLE`、`GLOBAL_BLOOM`。
### index_class_name [string]
@@ -100,11 +102,11 @@ import ChangeLog from '../changelog/connector-hudi.md';
### batch_interval_ms [Int]
-`batch_interval_ms` 批量写入 Hudi 表的时间间隔。
+`batch_interval_ms` 两次刷新到 Hudi 的最大时间间隔，单位为毫秒。
### batch_size [Int]
-`batch_size` 批量写入 Hudi 表的记录数大小.
+`batch_size` 单次刷新到 Hudi 前最多缓存的记录数。
### insert_shuffle_parallelism [Int]
@@ -135,25 +137,47 @@ import ChangeLog from '../changelog/connector-hudi.md';
`ERROR_WHEN_SCHEMA_NOT_EXIST`：当表不存在时将抛出错误<br/>
`IGNORE` ：忽略对表的处理<br/>
+### data_save_mode [Enum]
+
+在启动同步任务之前，针对目标端已有数据选择不同的处理方案：<br/>
+`DROP_DATA`：保留表结构并删除已有数据<br/>
+`APPEND_DATA`：保留表结构和已有数据<br/>
+`ERROR_WHEN_DATA_EXISTS`：当已有数据存在时报错<br/>
+
### 通用选项
Sink插件通用参数，请参考 [Sink Common Options](../common-options/sink-common-options.md)
了解详细信息。
## 示例
-### 单表
+### 单表 UPSERT
+
+当 `op_type` 为 `UPSERT` 时，必须配置 `record_key_fields`。
+
```hocon
sink {
Hudi {
- table_dfs_path = "hdfs://nameserivce/data/"
+ table_dfs_path = "/tmp/seatunnel_mnt/hudi"
database = "st"
- table_name = "test_table"
+ table_name = "st_test"
table_type = "COPY_ON_WRITE"
- conf_files_path =
"/home/test/hdfs-site.xml;/home/test/core-site.xml;/home/test/yarn-site.xml"
- batch_size = 10000
- use.kerberos = true
- kerberos.principal = "test_user@xxx"
- kerberos.principal.file = "/home/test/test_user.keytab"
+ op_type = "UPSERT"
+ record_key_fields = "c_bigint"
+ batch_size = 1000
+ batch_interval_ms = 1000
+ }
+}
+```
+
+### 最小单表配置
+
+追加写入时，通常只需要配置 `table_dfs_path` 和 `table_name`。
+
+```hocon
+sink {
+ Hudi {
+ table_dfs_path = "/tmp/seatunnel_mnt/hudi"
+ table_name = "st_test"
}
}
```
@@ -188,15 +212,14 @@ sink {
database = "st1"
table_name = "role"
table_type = "COPY_ON_WRITE"
- op_type="INSERT"
+ op_type = "INSERT"
batch_size = 10000
},
{
database = "st1"
table_name = "user"
table_type = "COPY_ON_WRITE"
- op_type="UPSERT"
- # op_type is 'UPSERT', must configured record_key_fields
+ op_type = "UPSERT"
record_key_fields = "user_id"
batch_size = 10000
},
@@ -206,7 +229,24 @@ sink {
table_type = "MERGE_ON_READ"
}
]
- ...
+ }
+}
+```
+
+### CDC 写入 Hudi
+
+当 Hudi 表需要保存 CDC 变更日志信息时，可以开启 `cdc_enabled`。
+
+```hocon
+sink {
+ Hudi {
+ table_dfs_path = "/tmp/seatunnel_mnt/hudi"
+ database = "st"
+ table_name = "st_test"
+ table_type = "COPY_ON_WRITE"
+ op_type = "UPSERT"
+ record_key_fields = "id"
+ cdc_enabled = true
}
}
```
diff --git a/docs/zh/connectors/sink/IoTDB.md b/docs/zh/connectors/sink/IoTDB.md
index db7988f9bd..427d09003d 100644
--- a/docs/zh/connectors/sink/IoTDB.md
+++ b/docs/zh/connectors/sink/IoTDB.md
@@ -19,11 +19,11 @@ import ChangeLog from '../changelog/connector-iotdb.md';
- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
> IoTDB 通过幂等写支持`精确一次`功能。如果两条数据使用相同的`key`和`timestamp`，新数据将覆盖旧数据。
-- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
+- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md)
## 支持的数据源信息
-| 数据源 | Supported 版本 | 地址 |
+| 数据源 | 支持的版本 | 地址 |
|-------|------------------------------|----------------|
| IoTDB | `0.13.0 <= version <= 1.3.X` | localhost:6667 |
@@ -49,18 +49,18 @@ import ChangeLog from '../changelog/connector-iotdb.md';
| password | String | 是 | -
| IoTDB 用户的密码
|
| key_device | String | 是 | -
| 在SeaTunnelRow中指定 IoTDB 设备ID的字段名
|
| key_timestamp | String | 否 | processing time
| 在SeaTunnelRow中指定 IoTDB 时间戳的字段名。如果未指定，则使用处理时间作为时间戳
|
-| key_measurement_fields | Array | 否 | exclude `device` &
`timestamp` | 在SeaTunnelRow中指定 IoTDB 测量列表的字段名称。如果未指定，则包括所有字段，但排除 `device` &
`timestamp` |
+| key_measurement_fields | Array | 否 | 排除设备和时间字段
| 在 SeaTunnelRow 中指定 IoTDB 测点字段列表。未配置时，会写入除 `key_device` 和 `key_timestamp`
对应字段外的其他字段 |
| storage_group | Array | 否 | -
| 指定设备存储组（路径前缀） <br/> 例如: deviceId = \${storage_group} + "." + \${key_device}
|
-| batch_size | Integer | 否 | 1024
| 对于批写入，当缓冲区的数量达到`batch_size`的数量或时间达到`batch_interval_ms`时，数据将被刷新到IoTDB中
|
-| max_retries | Integer | 否 | -
| 刷新的重试次数 failed
|
+| batch_size | Integer | 否 | 1024
| 批量写入时，当缓存行数达到 `batch_size`，数据会刷新到 IoTDB 中
|
+| max_retries | Integer | 否 | -
| 写入失败后的最大重试次数
|
| retry_backoff_multiplier_ms | Integer | 否 | -
| 用作生成下一个退避延迟的乘数
|
| max_retry_backoff_ms | Integer | 否 | -
| 尝试重试对 IoTDB 的请求之前等待的时间量
|
-| default_thrift_buffer_size | Integer | 否 | -
| 在 IoTDB 客户端中节省初始化缓冲区大小
|
-| max_thrift_frame_size | Integer | 否 | -
| 在 IoTDB 客户端中节约最大帧大小
|
-| zone_id | string | 否 | -
| IoTDB java.time.ZoneId client
|
+| default_thrift_buffer_size | Integer | 否 | -
| IoTDB 客户端的初始化 Thrift 缓冲区大小
|
+| max_thrift_frame_size | Integer | 否 | -
| IoTDB 客户端的最大 Thrift 帧大小
|
+| zone_id | string | 否 | -
| IoTDB 客户端使用的 `java.time.ZoneId`
|
| enable_rpc_compression | Boolean | 否 | -
| 在 IoTDB 客户端中启用rpc压缩
|
| connection_timeout_in_ms | Integer | 否 | -
| 连接到 IoTDB 时等待的最长时间（毫秒）
|
-| common-options | | 否 | -
| Sink 插件常用参数，详见 [Sink common
Options](../common-options/sink-common-options.md) |
+| common-options | | 否 | -
| Sink 插件通用参数，详见 [Sink Common
Options](../common-options/sink-common-options.md) |
## 示例
@@ -114,7 +114,7 @@ sink {
node_urls = ["localhost:6667"]
username = "root"
password = "root"
- key_device = "device_name" # specify the `deviceId` use device_name field
+ key_device = "device_name" # 使用 device_name 字段作为 deviceId
}
}
```
@@ -144,8 +144,8 @@ sink {
node_urls = ["localhost:6667"]
username = "root"
password = "root"
- key_device = "device_name" # specify the `deviceId` use device_name field
- key_timestamp = "event_ts" # specify the `timestamp` use event_ts field
+ key_device = "device_name" # 使用 device_name 字段作为 deviceId
+ key_timestamp = "event_ts" # 使用 event_ts 字段作为 timestamp
}
}
```
diff --git a/docs/zh/connectors/source/Easysearch.md
b/docs/zh/connectors/source/Easysearch.md
index 9b182bfc87..3e9b61d9bf 100644
--- a/docs/zh/connectors/source/Easysearch.md
+++ b/docs/zh/connectors/source/Easysearch.md
@@ -12,7 +12,7 @@ import ChangeLog from '../changelog/connector-easysearch.md';
## 描述
-用于从INFINI Easysearch读取数据。
+用于从 INFINI Easysearch 读取数据。
## 使用依赖
@@ -31,7 +31,9 @@ import ChangeLog from '../changelog/connector-easysearch.md';
支持的引擎
-* 支持发布的所有版本 [INFINI
Easysearch](https://www.infini.com/download/?product=easysearch).
+* 支持 [INFINI Easysearch](https://www.infini.com/download/?product=easysearch)
发布的所有版本。
+
+:::
## 数据类型映射
@@ -46,11 +48,32 @@ import ChangeLog from
'../changelog/connector-easysearch.md';
| LONG | LONG |
| FLOAT<br/>HALF_FLOAT | FLOAT |
| DOUBLE | DOUBLE |
-| Date | LOCAL_DATE_TIME_TYPE |
+| DATE | LOCAL_DATE_TIME_TYPE |
+
+## Source 选项
+
+| 名称 | 类型 | 是否必填 | 默认值 | 描述 |
+|------------------------|---------|----------|--------|------|
+| hosts | array | 是 | - | Easysearch HTTP 地址。 |
+| index | string | 是 | - | Easysearch 索引名，支持
`seatunnel-*` 这类通配符匹配。 |
+| username | string | 否 | - | 开启安全认证时使用的用户名。 |
+| password | string | 否 | - | 开启安全认证时使用的密码。 |
+| source | array | 否 | - | 要读取的索引字段。`source` 和
`schema` 二选一配置。 |
+| schema | config | 否 | - | SeaTunnel
读取和转换字段时使用的表结构。`schema` 和 `source` 二选一配置。 |
+| query | json | 否 | `{"match_all":{}}` | 用于过滤数据的
Easysearch DSL 查询。 |
+| scroll_time | string | 否 | 1m | Easysearch 保留 scroll
查询上下文的时间。 |
+| scroll_size | int | 否 | 100 | 每次 scroll 请求最多返回的数据条数。
|
+| tls_verify_certificate | boolean | 否 | true | 是否校验 HTTPS 证书。 |
+| tls_verify_hostname | boolean | 否 | true | 是否校验 HTTPS 主机名。 |
+| tls_keystore_path | string | 否 | - | PEM 或 JKS key store
文件路径。 |
+| tls_keystore_password | string | 否 | - | key store 密码。 |
+| tls_truststore_path | string | 否 | - | PEM 或 JKS trust store
文件路径。 |
+| tls_truststore_password | string | 否 | - | trust store 密码。 |
+| common-options | config | 否 | - | Source 插件通用参数。 |
### hosts [array]
-Easysearch集群http地址，格式为“host:port”，允许指定多个主机。例如`[“host1:9200”，“host2:9200”]`。
+Easysearch 集群 HTTP 地址，格式为 `host:port`，允许指定多个主机。例如 `["host1:9200",
"host2:9200"]`。
### username [string]
@@ -62,31 +85,30 @@ Easysearch集群http地址，格式为“host:port”，允许指定多个主机
### index [string]
-Easysearch搜索索引名称，支持*模糊匹配。
+Easysearch 索引名称，支持 `*` 通配符匹配。
### source [array]
索引字段。
-您可以通过指定字段“_id”来获取文档id。如果sink_id指向其他索引，由于Easysearch的限制，您需要为_id指定一个别名。
-若不配置源代码，则必须配置`schema`。
+可以通过指定字段 `_id` 来获取文档 id。如果要把 `_id` 写入其他索引，由于 Easysearch 限制，需要为 `_id` 指定别名。
+如果不配置 `source`，则必须配置 `schema`。
### query [json]
-Easysearch DSL.
-您可以控制读取数据的范围。
+Easysearch DSL 查询，用于控制读取数据的范围。
### scroll_time [String]
-Easysearch将为滚动请求保持搜索上下文活动的时间量。
+Easysearch 为 scroll 请求保留查询上下文的时间。
### scroll_size [int]
-每次Easysearch滚动请求返回的最大请求数。
+每次 Easysearch scroll 请求返回的最大数据条数。
### schema
-数据的结构，包括字段名和字段类型。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
-如果不配置schema，则必须配置`source`。
+数据结构，包括字段名和字段类型。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
+如果不配置 `schema`，则必须配置 `source`。
### tls_verify_certificate [boolean]
@@ -114,46 +136,55 @@ PEM或JKS信任存储的路径。运行SeaTunnel的操作系统用户必须能
### common options
-Source插件常用参数，详见[Source common
Options](../common-options/source-common-options.md)
+Source 插件通用参数，详见 [Source Common
Options](../common-options/source-common-options.md)。
## 示例
-简单的例子
+### 读取指定字段
```hocon
-Easysearch {
+source {
+ Easysearch {
hosts = ["localhost:9200"]
index = "seatunnel-*"
- source = ["_id","name","age"]
- query = {"range":{"firstPacket":{"gte":1700407367588,"lte":1700407367588}}}
+ source = ["_id", "name", "age"]
+ query = {"range": {"age": {"gte": 18, "lte": 60}}}
+ }
}
```
-复杂的例子
+### 使用 Schema 和 Query 读取
```hocon
-Easysearch {
- hosts = ["Easysearch:9200"]
+source {
+ Easysearch {
+ hosts = ["https://e2e_easysearch:9200";]
+ username = "admin"
+ password = "admin"
+ tls_verify_certificate = false
+ tls_verify_hostname = false
+
index = "st_index"
+ query = {"range": {"c_int": {"gte": 10, "lte": 20}}}
schema = {
- fields {
- c_map = "map<string, tinyint>"
- c_array = "array<tinyint>"
- c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
- c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(2, 1)"
- c_bytes = bytes
- c_date = date
- c_timestamp = timestamp
- }
+ fields {
+ c_map = "map<string, tinyint>"
+ c_array = "array<tinyint>"
+ c_string = string
+ c_boolean = boolean
+ c_tinyint = tinyint
+ c_smallint = smallint
+ c_int = int
+ c_bigint = bigint
+ c_float = float
+ c_double = double
+ c_decimal = "decimal(2, 1)"
+ c_bytes = bytes
+ c_date = date
+ c_timestamp = timestamp
+ }
}
- query = {"range":{"firstPacket":{"gte":1700407367588,"lte":1700407367588}}}
+ }
}
```
diff --git a/docs/zh/connectors/source/IoTDB.md
b/docs/zh/connectors/source/IoTDB.md
index 8270fb7ec0..8297633ce6 100644
--- a/docs/zh/connectors/source/IoTDB.md
+++ b/docs/zh/connectors/source/IoTDB.md
@@ -65,7 +65,7 @@ import ChangeLog from '../changelog/connector-iotdb.md';
| thrift_max_frame_size | int | 否 | - | Thrift 最大帧尺寸
|
| enable_cache_leader | boolean | 否 | - | 是否启用 Leader 节点缓存
|
| version | string | 否 | - | 客户端 SQL 语义版本（`V_0_12` /
`V_0_13`） |
-| common-options | | 否 | - | Source 插件常用参数，详见 [Source
common Options](../common-options/source-common-options.md) |
+| common-options | | 否 | - | Source 插件通用参数，详见 [Source
Common Options](../common-options/source-common-options.md) |
我们可以使用时间列进行分区查询。
@@ -110,6 +110,9 @@ source {
username = "root"
password = "root"
sql = "SELECT temperature, moisture, c_int, c_bigint, c_float, c_double,
c_string, c_boolean FROM root.test_group.* WHERE time < 4102329600000 align by
device"
+ lower_bound = 1
+ upper_bound = 4102329600000
+ num_partitions = 10
schema {
fields {
ts = timestamp
@@ -133,6 +136,8 @@ sink {
}
```
+`lower_bound`、`upper_bound` 和 `num_partitions`
是可选项。当查询覆盖很大的时间范围时，可以用它们把读取任务按时间范围拆成多个分区。
+
上游 IoTDB 的数据格式如下所示:
```shell
