Skip to content

[Doc] update iotdb document #5404

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
liugddx merged 3 commits into from
Sep 5, 2023
Merged

[Doc] update iotdb document #5404

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8cb40f7
[Doc] update iotdb document, remove duplicated config
liunaijie Aug 30, 2023
006147b
[Doc] update iotdb document
liunaijie Sep 4, 2023
26c31dd
[Doc] update iotdb document
liunaijie Sep 4, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
110 changes: 75 additions & 35 deletions docs/en/connector-v2/sink/IoTDB.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,10 +50,10 @@ There is a conflict of thrift version between IoTDB and Spark.Therefore, you nee

| Name | Type | Required | Default | Description |
|-----------------------------|---------|----------|--------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| node_urls | Array | Yes | - | `IoTDB` cluster address, the format is `["host:port", ...]` |
| node_urls | String | Yes | - | `IoTDB` cluster address, the format is `"host1:port"` or `"host1:port,host2:port"` |
| username | String | Yes | - | `IoTDB` user username |
| password | String | Yes | - | `IoTDB` user password |
| key_device | String | No | - | Specify field name of the `IoTDB` deviceId in SeaTunnelRow |
| 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` |
| storage_group | Array | No | - | Specify device storage group(path prefix) <br/> example: deviceId = ${storage_group} + "." + ${key_device} |
Expand All @@ -68,78 +68,118 @@ There is a conflict of thrift version between IoTDB and Spark.Therefore, you nee
| connection_timeout_in_ms | Integer | No | - | The maximum time (in ms) to wait when connecting to `IoTDB` |
| common-options | | no | - | Sink plugin common parameters, please refer to [Sink Common Options](common-options.md) for details |

## Task Example
## Examples

```hocon
env {
execution.parallelism = 2
job.mode = "BATCH"
}

source {
FakeSource {
row.num = 16
bigint.template = [1664035200001]
schema = {
fields {
device_name = "string"
temperature = "float"
moisture = "int"
event_ts = "bigint"
field_1 = "string"
field_2 = "string"
Comment thread
liunaijie marked this conversation as resolved.
Outdated
}
}
}
}

...

```

Upstream SeaTunnelRow data format is the following:

| device_name | temperature | moisture | event_ts | field_1 | field_2 |
|--------------------------|-------------|----------|---------------|---------|---------|
| root.test_group.device_a | 36.1 | 100 | 1664035200001 | abc1 | def1 |
| root.test_group.device_b | 36.2 | 101 | 1664035200001 | abc2 | def2 |
| root.test_group.device_c | 36.3 | 102 | 1664035200001 | abc3 | def3 |

### Case1

Common options:
only fill required config, use current processing time as timestamp. and include all fields but exclude `device` & `timestamp` as measurement fields

```hocon
sink {
IoTDB {
node_urls = ["localhost:6667"]
node_urls = "localhost:6667"
username = "root"
password = "root"
batch_size = 1024
key_device = "device_name"
Comment thread
liunaijie marked this conversation as resolved.
Outdated
}
}
```

When you assign `key_device` is `device_name`, for example:
Output to `IoTDB` data format is the following:

```shell
IoTDB> SELECT * FROM root.test_group.* align by device;
+------------------------+------------------------+--------------+-----------+---------------+---------+---------+
| Time| Device| temperature| moisture| event_ts | field_1 | field_2 |
+------------------------+------------------------+--------------+-----------+---------------+---------+---------+
|2023-09-01T00:00:00.001Z|root.test_group.device_a| 36.1| 100| 1664035200001 | abc1 | def1 |
|2023-09-01T00:00:00.001Z|root.test_group.device_b| 36.2| 101| 1664035200001 | abc2 | def2 |
|2023-09-01T00:00:00.001Z|root.test_group.device_c| 36.3| 102| 1664035200001 | abc3 | def3 |
+------------------------+------------------------+--------------+-----------+---------------+---------+---------+
```

### Case2

use source event's time

```hocon
sink {
IoTDB {
...
node_urls = "localhost:6667"
username = "root"
password = "root"
key_device = "device_name"
key_timestamp = "event_ts"
Comment thread
liunaijie marked this conversation as resolved.
Outdated
key_measurement_fields = ["temperature", "moisture"]
}
}
```

Upstream SeaTunnelRow data format is the following:

@hailin0 hailin0 Aug 31, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why delete？

@liunaijie liunaijie Aug 31, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the sink example, I only add 1 case (fakesource to iotdb)(also 1 case in source example, use iotdb to console)

the key_timestamp is optional, so i remove it.
for example data, remove some filed to keep consistent with example code.


| device_name | field_1 | field_2 |
|--------------------------|---------|---------|
| root.test_group.device_a | 1001 | 1002 |
| root.test_group.device_b | 2001 | 2002 |
| root.test_group.device_c | 3001 | 3002 |

Output to `IoTDB` data format is the following:

```shell
IoTDB> SELECT * FROM root.test_group.* align by device;
+------------------------+------------------------+-----------+----------+
| Time| Device| field_1| field_2|
+------------------------+------------------------+----------+-----------+
|2022-09-26T17:50:01.201Z|root.test_group.device_a| 1001| 1002|
|2022-09-26T17:50:01.202Z|root.test_group.device_b| 2001| 2002|
|2022-09-26T17:50:01.203Z|root.test_group.device_c| 3001| 3002|
+------------------------+------------------------+----------+-----------+
+------------------------+------------------------+--------------+-----------+
| Time| Device| temperature| moisture|
+------------------------+------------------------+--------------+-----------+
|2022-09-25T00:00:00.001Z|root.test_group.device_a| 36.1| 100|
|2022-09-25T00:00:00.001Z|root.test_group.device_b| 36.2| 101|
|2022-09-25T00:00:00.001Z|root.test_group.device_c| 36.3| 102|
+------------------------+------------------------+--------------+-----------+
```

### Case2
### Case3

When you assign `key_device`、`key_timestamp`、`key_measurement_fields`, for example:
use source event's time and limit measurement fields

```hocon
sink {
IoTDB {
...
node_urls = "localhost:6667"
username = "root"
password = "root"
key_device = "device_name"
key_timestamp = "ts"

@hailin0 hailin0 Aug 31, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why delete？

key_timestamp = "event_ts"
key_measurement_fields = ["temperature", "moisture"]
}
}
```

Upstream SeaTunnelRow data format is the following:

| ts | device_name | field_1 | field_2 | temperature | moisture |

@hailin0 hailin0 Aug 31, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

whl update ？

|---------------|--------------------------|---------|---------|-------------|----------|
| 1664035200001 | root.test_group.device_a | 1001 | 1002 | 36.1 | 100 |
| 1664035200001 | root.test_group.device_b | 2001 | 2002 | 36.2 | 101 |
| 1664035200001 | root.test_group.device_c | 3001 | 3002 | 36.3 | 102 |

Output to `IoTDB` data format is the following:

```shell
Expand Down
187 changes: 65 additions & 122 deletions docs/en/connector-v2/source/IoTDB.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,16 @@

> IoTDB source connector

## Description
## Support Those Engines

Read external data source data through IoTDB.
> Spark<br/>
> Flink<br/>
> SeaTunnel Zeta<br/>

## Key features

- [x] [batch](../../concept/connector-v2-features.md)
- [ ] [stream](../../concept/connector-v2-features.md)
- [x] [stream](../../concept/connector-v2-features.md)
- [x] [exactly-once](../../concept/connector-v2-features.md)
- [x] [column projection](../../concept/connector-v2-features.md)

Expand All @@ -18,106 +20,53 @@ supports query SQL and can achieve projection effect.
- [x] [parallelism](../../concept/connector-v2-features.md)
- [ ] [support user-defined split](../../concept/connector-v2-features.md)

## Options

| name | type | required | default value |
|----------------------------|---------|----------|---------------|
| host | string | no | - |
| port | int | no | - |
| node_urls | string | no | - |
| username | string | yes | - |
| password | string | yes | - |
| sql | string | yes | - |
| schema | config | yes | - |
| fetch_size | int | no | - |
| lower_bound | long | no | - |
| upper_bound | long | no | - |
| num_partitions | int | no | - |
| thrift_default_buffer_size | int | no | - |
| enable_cache_leader | boolean | no | - |
| version | string | no | - |
| common-options | | no | - |

### single node, you need to set host and port to connect to the remote data source.

**host** [string] the host of the IoTDB when you select host of the IoTDB

**port** [int] the port of the IoTDB when you select

### multi node, you need to set node_urls to connect to the remote data source.

**node_urls** [string] the node_urls of the IoTDB when you select

e.g.

```text
127.0.0.1:8080,127.0.0.2:8080
```

### other parameters

**sql** [string]
execute sql statement e.g.

```
select name,age from test
```

### schema [config]

#### fields [Config]

The schema of the IoTDB that you want to generate

e.g.

```
schema {
fields {
name = string
age = int
}
}
```

### option parameters

### fetch_size [int]

the fetch_size of the IoTDB when you select

### username [string]

the username of the IoTDB when you select

### password [string]

the password of the IoTDB when you select

### lower_bound [long]

the lower_bound of the IoTDB when you select

### upper_bound [long]

the upper_bound of the IoTDB when you select

### num_partitions [int]

the num_partitions of the IoTDB when you select

### thrift_default_buffer_size [int]

the thrift_default_buffer_size of the IoTDB when you select

### enable_cache_leader [boolean]

enable_cache_leader of the IoTDB when you select
## Description

### version [string]
Read external data source data through IoTDB.

Version represents the SQL semantic version used by the client, which is used to be compatible with the SQL semantics of
0.12 when upgrading 0.13. The possible values are: V_0_12, V_0_13.
:::tip

There is a conflict of thrift version between IoTDB and Spark.Therefore, you need to execute `rm -f $SPARK_HOME/jars/libthrift*` and `cp $IOTDB_HOME/lib/libthrift* $SPARK_HOME/jars/` to resolve it.

:::

## Supported DataSource Info

| Datasource | Supported Versions | Url |
|------------|--------------------|----------------|
| IoTDB | `>= 0.13.0` | localhost:6667 |

## Data Type Mapping

| IotDB Data type | SeaTunnel Data type |
|-----------------|---------------------|
| BOOLEAN | BOOLEAN |
| INT32 | TINYINT |
| INT32 | SMALLINT |
| INT32 | INT |
| INT64 | BIGINT |
| FLOAT | FLOAT |
| DOUBLE | DOUBLE |
| TEXT | STRING |

## Source Options

| name | type | required | default value | Description |
Comment thread
liunaijie marked this conversation as resolved.
Outdated
|----------------------------|---------|----------|---------------|------------------------------------------------------------------------------------|
| node_urls | string | yes | - | `IoTDB` cluster address, the format is `"host1:port"` or `"host1:port,host2:port"` |
| username | string | yes | - | `IoTDB` user username |
| password | string | yes | - | `IoTDB` user password |
| sql | string | yes | - | execute sql statement |
| schema | config | yes | - | the data schema |
| 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 |
| common-options | | no | - | |

### split partitions

Expand Down Expand Up @@ -157,37 +106,31 @@ Source plugin common parameters, please refer to [Source Common Options](common-

## Examples

### Case1

Common options:

```hocon
env {
execution.parallelism = 2
job.mode = "BATCH"
}

source {
IoTDB {
node_urls = "localhost:6667"
username = "root"
password = "root"
sql = "SELECT temperature, moisture FROM root.test_group.* WHERE time < 4102329600000 align by device"
schema {
fields {
ts = timestamp
device_name = string
temperature = float
moisture = bigint
Comment thread
liunaijie marked this conversation as resolved.
}
}
}
}
```

When you assign `sql`、`fields`、`partition`, for example:

```hocon
sink {
IoTDB {
...
sql = "SELECT temperature, moisture FROM root.test_group.* WHERE time < 4102329600000 align by device"
lower_bound = 1
upper_bound = 4102329600000
num_partitions = 10
fields {
ts = bigint
device_name = string

temperature = float
moisture = bigint
}
Console {
}
}
```
Expand Down
Loading