| <!-- |
| |
| Licensed to the Apache Software Foundation (ASF) under one |
| or more contributor license agreements. See the NOTICE file |
| distributed with this work for additional information |
| regarding copyright ownership. The ASF licenses this file |
| to you under the Apache License, Version 2.0 (the |
| "License"); you may not use this file except in compliance |
| with the License. You may obtain a copy of the License at |
| |
| http://www.apache.org/licenses/LICENSE-2.0 |
| |
| Unless required by applicable law or agreed to in writing, |
| software distributed under the License is distributed on an |
| "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| KIND, either express or implied. See the License for the |
| specific language governing permissions and limitations |
| under the License. |
| |
| --> |
| |
| |
| # Apache SeaTunnel |
| |
| ## 1. Overview |
| |
| SeaTunnel is a distributed integration platform designed for massive data. Leveraging its high performance and elastic scaling capabilities, it connects multi-source heterogeneous data links through standardized Connectors (composed of Source and Sink). The platform uniformly abstracts various data sources into the SeaTunnelRow format via Source. After dynamic resource scheduling and batch processing optimization, it efficiently writes data to different storage systems through Sink. Through the deep integration of the IoTDB Connector with SeaTunnel, it not only addresses core challenges in time-series data scenarios such as **high-throughput writing, multi-source governance, and complex analysis**, but also helps enterprises quickly build **low-cost, highly reliable, and easily scalable** data infrastructure in fields like the Internet of Things and industrial internet, leveraging the out-of-the-box connector ecosystem and automated operation and maintenance capabilities. |
| |
| ## 2. Usage Steps |
| |
| ### 2.1 Environment Preparation |
| |
| #### 2.1.1 Software Requirements |
| |
| | Software | Version | Installation Reference | |
| | ------------- | ------------- | ----------------------------------------------------- | |
| | IoTDB | >= 2.0.5 | [Quick Start](../QuickStart/QuickStart_apache.md) | |
| | SeaTunnel | 2.3.12 | [Official Website](https://seatunnel.apache.org/download) | |
| |
| * Thrift Version Conflict Resolution (Only required for Spark engine): |
| |
| ```Bash |
| # Remove older Thrift from Spark |
| rm -f $SPARK_HOME/jars/libthrift* |
| # Copy IoTDB's Thrift library to Spark classpath |
| cp $IOTDB_HOME/lib/libthrift* $SPARK_HOME/jars/ |
| ``` |
| |
| #### 2.1.2 Dependency Configuration |
| |
| 1. JDBC |
| |
| * Spark/Flink Engine: Place the [JDBC driver JAR](https://mvnrepository.com/artifact/org.apache.iotdb/iotdb-jdbc) into the `${SEATUNNEL_HOME}/plugins/` directory. |
| * SeaTunnel Zeta Engine: Place the [JDBC driver JAR](https://mvnrepository.com/artifact/org.apache.iotdb/iotdb-jdbc) into the `${SEATUNNEL_HOME}/lib/` directory. |
| |
| 2. Connector |
| |
| Place the corresponding version of the [SeaTunnel Connector](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-iotdb) into the `${SEATUNNEL_HOME}/plugins/` directory. |
| |
| ### 2.2 Reading Data (IoTDB Source Connector) |
| |
| #### 2.2.1 Configuration Parameters |
| |
| | **Parameter** | **Type** | **Required** | **Default** | **Description** | |
| | -------------------------- | -------- | ------------ | ----------- | --------------------------------------------------------------------------------------------------------------------------------------- | |
| | `node_urls` | string | yes | - | IoTDB cluster address, format: `"host1:port"` or `"host1:port,host2:port"` | |
| | `username` | string | yes | - | IoTDB username | |
| | `password` | string | yes | - | IoTDB password | |
| | `sql_dialect` | string | no | tree | IoTDB model: `tree` for tree model; `table` for table model | |
| | `sql` | string | yes | - | SQL query statement to execute | |
| | `database` | string | no | - | Database name, only effective in table model | |
| | `schema` | config | yes | - | Data schema definition | |
| | `fetch_size` | int | no | - | Number of data rows fetched per request from IoTDB during query execution | |
| | `lower_bound` | long | no | - | Lower bound of time range (used for data partitioning by time column) | |
| | `upper_bound` | long | no | - | Upper bound of time range (used for data partitioning by time column) | |
| | `num_partitions` | int | no | - | Number of partitions (used when partitioning by time column): <br> 1 partition: uses the full time range <br> If partitions < (upper_bound - lower_bound), the difference is used as actual partitions | |
| | `thrift_default_buffer_size`| int | no | - | Thrift protocol buffer size | |
| | `thrift_max_frame_size` | int | no | - | Thrift maximum frame size | |
| | `enable_cache_leader` | boolean | no | - | Whether to enable leader node caching | |
| | `version` | string | no | - | Client SQL semantic version (`V_0_12` / `V_0_13`) | |
| |
| #### 2.2.2 Configuration Example |
| |
| 1. Create a new file `iotdb_source_example.conf` in the `${SEATUNNEL_HOME}/config/` directory: |
| |
| ```bash |
| env { |
| parallelism = 2 # Parallelism set to 2 |
| job.mode = "BATCH" # Batch mode |
| } |
| |
| source { |
| IoTDB { |
| node_urls = "localhost:6667" |
| username = "root" |
| password = "root" |
| sql = "SELECT temperature, humidity, status FROM root.testdb.seatunnel.source.device align by device" |
| schema { |
| fields { |
| ts = timestamp |
| device_name = string |
| temperature = double |
| humidity = double |
| status = boolean |
| } |
| } |
| } |
| } |
| |
| sink { |
| Console { |
| } # Output to console |
| } |
| ``` |
| |
| 2. Run SeaTunnel with the following command: |
| |
| ```Bash |
| ./bin/seatunnel.sh --config config/iotdb_source_example.conf -e local |
| ``` |
| |
| 3. For more details, please refer to the official Apache SeaTunnel documentation on [IoTDB Source Connector](https://seatunnel.apache.org/docs/2.3.12/connector-v2/source/IoTDB). |
| |
| ### 2.3 Writing Data (IoTDB Sink Connector) |
| |
| #### 2.3.1 Configuration Parameters |
| |
| | **Parameter** | **Type** | **Required** | **Default** | **Description** | |
| | ----------------------------- | --------- | ------------ | ------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | |
| | `node_urls` | Array | yes | - | IoTDB cluster address, format: `["host1:port"]` or `["host1:port","host2:port"]` | |
| | `username` | String | yes | - | IoTDB username | |
| | `password` | String | yes | - | IoTDB password | |
| | `sql_dialect` | String | no | tree | IoTDB model: `tree` for tree model; `table` for table model | |
| | `storage_group` | String | yes | - | IoTDB tree model: specifies the storage group for devices (path prefix) e.g., deviceId = \${storage_group} + "." + \${key_device}; IoTDB table model: specifies the database | |
| | `key_device` | String | yes | - | IoTDB tree model: field name in SeaTunnelRow that specifies the IoTDB device ID; IoTDB table model: field name in SeaTunnelRow that specifies the IoTDB table name | |
| | `key_timestamp` | String | no | processing time | IoTDB tree model: field name in SeaTunnelRow that specifies the IoTDB timestamp (if not specified, processing time is used as timestamp); IoTDB table model: field name in SeaTunnelRow that specifies the IoTDB time column (if not specified, processing time is used as timestamp) | |
| | `key_measurement_fields` | Array | no | See description | IoTDB tree model: field names in SeaTunnelRow that specify the list of IoTDB measurements (if not specified, includes all fields except `key_device` and `key_timestamp`); IoTDB table model: field names in SeaTunnelRow that specify the IoTDB field columns (if not specified, includes all fields except `key_device`, `key_timestamp`, `key_tag_fields`, `key_attribute_fields`) | |
| | `key_tag_fields` | Array | no | - | IoTDB tree model: not applicable; IoTDB table model: field names in SeaTunnelRow that specify the IoTDB tag columns | |
| | `key_attribute_fields` | Array | no | - | IoTDB tree model: not applicable; IoTDB table model: field names in SeaTunnelRow that specify the IoTDB attribute columns | |
| | `batch_size` | Integer | no | 1024 | For batch writing, data is flushed to IoTDB when the buffer reaches `batch_size` or when the time reaches `batch_interval_ms` | |
| | `max_retries` | Integer | no | - | Number of retries on failed flush | |
| | `retry_backoff_multiplier_ms` | Integer | no | - | Multiplier used to generate the next backoff delay | |
| | `max_retry_backoff_ms` | Integer | no | - | Maximum wait time before retrying a request to IoTDB | |
| | `default_thrift_buffer_size` | Integer | no | - | Initial buffer size for Thrift client in IoTDB | |
| | `max_thrift_frame_size` | Integer | no | - | Maximum frame size for Thrift client in IoTDB | |
| | `zone_id` | string | no | - | IoTDB client `java.time.ZoneId` | |
| | `enable_rpc_compression` | Boolean | no | - | Enable RPC compression in IoTDB client | |
| | `connection_timeout_in_ms` | Integer | no | - | Maximum time (in milliseconds) to wait when connecting to IoTDB | |
| |
| #### 2.3.2 Configuration Example |
| |
| 1. Create a new file `iotdb_sink_example.conf` in the `${SEATUNNEL_HOME}/config/` directory: |
| |
| ```bash |
| # Define runtime environment |
| env { |
| parallelism = 4 |
| job.mode = "BATCH" |
| } |
| |
| source { |
| Jdbc { |
| url = "jdbc:mysql://localhost:3306/demo_db?useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true" |
| driver = "com.mysql.cj.jdbc.Driver" |
| connection_check_timeout_sec = 100 |
| user = "root" |
| password = "IoTDB@2024" |
| query = "select * from device" |
| } |
| } |
| |
| sink { |
| IoTDB { |
| node_urls = ["localhost:6667"] |
| username = "root" |
| password = "root" |
| key_device = "id" # Specify the `deviceId` using the device_name field |
| key_timestamp = "intime" |
| storage_group = "root.mysql" |
| } |
| } |
| ``` |
| |
| 2. Run SeaTunnel with the following command: |
| |
| ```Bash |
| ./bin/seatunnel.sh --config config/iotdb_sink_example.conf -e local |
| ``` |
| |
| 3. For more configuration parameters and examples, please refer to the official Apache SeaTunnel documentation on [IoTDB Sink Connector](https://seatunnel.apache.org/docs/2.3.12/connector-v2/sink/IoTDB). |
