Version: Next

IoTDBv2

IoTDBv2 sink connector

Support Those Engines

Spark
Flink
SeaTunnel Zeta

Description

Used to write data to IoTDB 2.x. The connector name in job configuration is IoTDBv2.

Key Features

exactly-once
IoTDB supports the exactly-once feature through idempotent writing. If multiple data have the same key and timestamp, the latest one will overwrite the previous one.
timer flush

Supported DataSource Info

Datasource	Supported Versions	Url
IoTDB	`2.0 <= version`	localhost:6667

Data Type Mapping

SeaTunnel Data Type	IoTDB Data Type
BOOLEAN	BOOLEAN
TINYINT	INT32
SMALLINT	INT32
INT	INT32
BIGINT	INT64
FLOAT	FLOAT
DOUBLE	DOUBLE
STRING	STRING
TIMESTAMP	TIMESTAMP
DATE	DATE

Sink Options

Name	Type	Required	Default	Description
node_urls	Array	Yes	-	IoTDB cluster address, the format is `["host1:port"]` or `["host1:port","host2:port"]`
username	String	Yes	-	IoTDB username
password	String	Yes	-	IoTDB user password
sql_dialect	String	No	tree	The SQL dialect of IoTDB. Available values are `"tree"` and `"table"`.
storage_group	String	Yes	-	IoTDB tree model: device path prefix. For example, the device path is `storage_group + "." + key_device`. Set it to an empty string when `key_device` already contains the full device path. IoTDB table model: database name.
key_device	String	Yes	-	IoTDB-tree: Specify the field name in SeaTunnelRow to be used as device id IoTDB-table: Specify the field name in SeaTunnelRow to be used as table name
key_timestamp	String	No	processing time	IoTDB-tree: Specify the field name in SeaTunnelRow to be used as timestamp (processing time will be used by default) IoTDB-table: Specify the field name in SeaTunnelRow to be used as time column (processing time will be used by default)
key_measurement_fields	Array	No	refer to description	IoTDB-tree: Specify the field names in SeaTunnelRow to be used as measurement (all fields excluding `key_device`&`key_timestamp` will be used by default) IoTDB-table: Specify the field names in SeaTunnelRow to be used as FIELD columns (all fields excluding `key_device`, `key_timestamp`, `key_tag_fields` and `key_attribute_fields` will be used by default)
key_tag_fields	Array	No	-	IoTDB-tree: invalid IoTDB-table: Specify the field names in SeaTunnelRow to be used as TAG columns
key_attribute_fields	Array	No	-	IoTDB-tree: invalid IoTDB-table: Specify the field names in SeaTunnelRow to be used as ATTRIBUTE columns
batch_size	Integer	No	1024	The connector flushes buffered records to IoTDB when the number of buffered records reaches `batch_size`. It also flushes before checkpoint commit and when the writer is closed.
max_retries	Integer	No	0	The maximum number of retries when flushing records fails.
retry_backoff_multiplier_ms	Integer	No	0	The multiplier used to calculate the retry backoff delay.
max_retry_backoff_ms	Integer	No	0	The maximum retry backoff delay in milliseconds.
default_thrift_buffer_size	Integer	No	-	Thrift init buffer size in IoTDB client
max_thrift_frame_size	Integer	No	-	Thrift max frame size in IoTDB client
zone_id	String	No	-	java.time.ZoneId in IoTDB client
enable_rpc_compression	Boolean	No	-	Enable rpc compression in IoTDB client, only valid in IoTDB-tree
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 for details

For the tree model, key_device is used as the IoTDB device path, and key_measurement_fields controls which fields become measurements. If key_measurement_fields is not set, all fields except key_device and key_timestamp are written as measurements.

For the table model, storage_group is the database, key_device is the target table name field, key_tag_fields are TAG columns, key_attribute_fields are ATTRIBUTE columns, and key_measurement_fields are FIELD columns. If key_measurement_fields is not set, all fields except the table name, time, TAG, and ATTRIBUTE fields are written as FIELD columns.

Examples

Example 1: Write data to IoTDB-tree

env {
  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"
        c_string = "string"
        c_boolean = "boolean"
        c_tinyint = "tinyint"
        c_smallint = "smallint"
        c_int = "int"
        c_bigint = "bigint"
        c_float = "float"
        c_double = "double"
      }
    }
  }
}

The data format from upstream SeaTunnelRow is as follows:

device_name	temperature	moisture	event_ts	c_string	c_boolean	c_tinyint	c_smallint	c_int	c_bigint	c_float	c_double
root.test_group.device_a	36.1	100	1664035200001	abc1	true	1	1	1	2147483648	1.0	1.0
root.test_group.device_b	36.2	101	1664035200001	abc2	false	2	2	2	2147483649	2.0	2.0
root.test_group.device_c	36.3	102	1664035200001	abc3	false	3	3	3	2147483649	3.0	3.0

Case 1

Only required options used:

use current processing time as timestamp
measurement fields include all fields excluding key_device

sink {
  IoTDBv2 {
    node_urls = ["localhost:6667"]
    username = "root"
    password = "root"
    key_device = "device_name" # specify the `deviceId` use device_name field
  }
}

The data format of IoTDB output is as follows:

IoTDB> SELECT * FROM root.test_group.* align by device;
+------------------------+------------------------+--------------+-----------+--------------+---------+----------+----------+-----------+------+-----------+--------+---------+
|                    Time|                  Device|   temperature|   moisture|      event_ts| c_string| c_boolean| c_tinyint| c_smallint| c_int|   c_bigint| c_float| c_double|
+------------------------+------------------------+--------------+-----------+--------------+---------+----------+----------+-----------+------+-----------+--------+---------+
|2023-09-01T00:00:00.001Z|root.test_group.device_a|          36.1|        100| 1664035200001|     abc1|      true|         1|          1|     1| 2147483648|     1.0|      1.0| 
|2023-09-01T00:00:00.001Z|root.test_group.device_b|          36.2|        101| 1664035200001|     abc2|     false|         2|          2|     2| 2147483649|     2.0|      2.0|
|2023-09-01T00:00:00.001Z|root.test_group.device_c|          36.3|        102| 1664035200001|     abc2|     false|         3|          3|     3| 2147483649|     3.0|      3.0|
+------------------------+------------------------+--------------+-----------+--------------+---------+---------+-----------+-----------+------+-----------+--------+---------+

Case 2

Use source event's time:

use key_timestamp as timestamp
measurement fields include all fields excluding key_device & key_timestamp

sink {
  IoTDBv2 {
    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
  }
}

The data format of IoTDB output is as follows:

IoTDB> SELECT * FROM root.test_group.* align by device;
+------------------------+------------------------+--------------+-----------+--------------+---------+----------+----------+-----------+------+-----------+--------+---------+
|                    Time|                  Device|   temperature|   moisture|      event_ts| c_string| c_boolean| c_tinyint| c_smallint| c_int|   c_bigint| c_float| c_double|
+------------------------+------------------------+--------------+-----------+--------------+---------+----------+----------+-----------+------+-----------+--------+---------+
|2022-09-25T00:00:00.001Z|root.test_group.device_a|          36.1|        100| 1664035200001|     abc1|      true|         1|          1|     1| 2147483648|     1.0|      1.0| 
|2022-09-25T00:00:00.001Z|root.test_group.device_b|          36.2|        101| 1664035200001|     abc2|     false|         2|          2|     2| 2147483649|     2.0|      2.0|
|2022-09-25T00:00:00.001Z|root.test_group.device_c|          36.3|        102| 1664035200001|     abc2|     false|         3|          3|     3| 2147483649|     3.0|      3.0|
+------------------------+------------------------+--------------+-----------+--------------+---------+---------+-----------+-----------+------+-----------+--------+---------+

Case 3

Use source event's time and limit measurement fields:

use key_timestamp as timestamp
measurement fields include only fields specified in key_measurement_fields

sink {
  IoTDBv2 {
    node_urls = ["localhost:6667"]
    username = "root"
    password = "root"
    key_device = "device_name"
    key_timestamp = "event_ts"
    key_measurement_fields = ["temperature", "moisture"]
  }
}

The data format of IoTDB output is as follows:

IoTDB> SELECT * FROM root.test_group.* align by device;
+------------------------+------------------------+--------------+-----------+
|                    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|
+------------------------+------------------------+--------------+-----------+

Example 2： Write data into IoTDB-table

env {
  parallelism = 2
  job.mode = "BATCH"
}

source {
  FakeSource {
    ...
    schema = {
      fields {
        ts = timestamp
        model_id = string
        region = string
        tag = string
        status = boolean
        arrival_date = date
        temperature = double
      }
    }
  }
}

The data format from upstream SeaTunnelRow is as follows:

ts	model_id	region	tag	status	arrival_date	temperature
2025-07-30T17:52:34.851	id1	0700HK	tag1	true	2024-11-12	4.34
2025-07-29T17:51:34.851	id2	0700HK	tag2	false	2024-12-01	5.54
2025-07-28T17:50:34.851	id3	0700HK	tag3	false	2024-12-22	7.34

Case 1

Only required options used:

use current processing time as timestamp
FIELD columns include all fields excluding key_device

sink {
  IoTDBv2 {
    node_urls = ["localhost:6667"]
    username = "root"
    password = "root"
    sql_dialect = "table"
    storage_group = "test_database"
    key_device = "region" 
  }
}

The data format of IoTDB output is as follows:

IoTDB> SELECT * FROM "test_database"."0700HK";
+-----------------------------+-----------------------+--------+----+------+------------+-----------+
|                         time|                     ts|model_id| tag|status|arrival_date|temperature|
+-----------------------------+-----------------------+--------+----+------+------------+-----------+
|2025-08-14T17:52:34.851+08:00|2025-07-30T17:52:34.851|     id1|tag1|  true|  2024-11-12|       4.34|
|2025-08-14T17:51:34.851+08:00|2025-07-29T17:51:34.851|     id2|tag2| false|  2024-12-01|       5.54|
|2025-08-14T17:50:34.851+08:00|2025-07-28T17:50:34.851|     id3|tag3| false|  2024-12-22|       7.34|
+-----------------------------+-----------------------+--------+----+------+------------+-----------+

IoTDB> DESC "test_database"."0700HK";
+------------+---------+--------+
|  ColumnName| DataType|Category|
+------------+---------+--------+
|        time|TIMESTAMP|    TIME|
|          ts|TIMESTAMP|   FIELD|
|    model_id|   STRING|   FIELD|
|         tag|   STRING|   FIELD|
|      status|  BOOLEAN|   FIELD|
|arrival_date|     DATE|   FIELD|
| temperature|   DOUBLE|   FIELD|
+------------+---------+--------+

Case 2

Use source event's time and limit TAG and ATTRIBUTE columns:

use key_timestamp as time column
use specified fields as TAG columns and ATTRIBUTE columns
FIELD columns include all fields excluding key_device,key_timestamp,key_tag_fieldsandkey_attribute_fields

sink {
  IoTDBv2 {
    node_urls = ["localhost:6667"]
    username = "root"
    password = "root"
    sql_dialect = "table"
    storage_group = "test_database"
    key_device = "region" 
    key_timestamp = "ts"
    key_tag_fields = ["tag"]
    key_attribute_fields = ["model_id"]
  }
}

The data format of IoTDB output is as follows:

IoTDB> SELECT * FROM "test_database"."0700HK";
+-----------------------------+----+--------+------+------------+-----------+
|                         time| tag|model_id|status|arrival_date|temperature|
+-----------------------------+----+--------+------+------------+-----------+
|2025-07-30T17:52:34.851+08:00|tag1|     id1|  true|  2024-11-12|       4.34|
|2025-07-29T17:51:34.851+08:00|tag2|     id2| false|  2024-12-01|       5.54|
|2025-07-28T17:50:34.851+08:00|tag3|     id3| false|  2024-12-22|       7.34|
+-----------------------------+----+--------+------+------------+-----------+

IoTDB> DESC "test_database"."0700HK";
+------------+---------+---------+
|  ColumnName| DataType| Category|
+------------+---------+---------+
|        time|TIMESTAMP|     TIME|
|         tag|   STRING|      TAG|
|    model_id|   STRING|ATTRIBUTE|
|      status|  BOOLEAN|    FIELD|
|arrival_date|     DATE|    FIELD|
| temperature|   DOUBLE|    FIELD|
+------------+---------+---------+

Case 3

Use source event's time and limit FIELD columns:

use key_timestamp as time column
use specified fields as FIELD columns

sink {
  IoTDBv2 {
    node_urls = ["localhost:6667"]
    username = "root"
    password = "root"
    sql_dialect = "table"
    storage_group = "test_database"
    key_device = "region" 
    key_timestamp = "ts"
    key_measurement_fields = ["status", "temperature"]
  }
}

The data format of IoTDB output is as follows:

IoTDB> SELECT * FROM "test_database"."0700HK";
+-----------------------------+------+-----------+
|                         time|status|temperature|
+-----------------------------+------+-----------+
|2025-07-30T17:52:34.851+08:00|  true|       4.34|
|2025-07-29T17:51:34.851+08:00| false|       5.54|
|2025-07-28T17:50:34.851+08:00| false|       7.34|
+-----------------------------+------+-----------+

IoTDB> DESC "test_database"."0700HK";
+-----------+---------+--------+
| ColumnName| DataType|Category|
+-----------+---------+--------+
|       time|TIMESTAMP|    TIME|
|     status|  BOOLEAN|   FIELD|
|temperature|   DOUBLE|   FIELD|
+-----------+---------+-------+

Changelog

Change Log

Change	Commit	Version
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)	https://github.com/apache/seatunnel/commit/4f5adeb1c7	2.3.11
[improve] iotdb options (#8965)	https://github.com/apache/seatunnel/commit/6e073935f4	2.3.10
[Improve] restruct connector common options (#8634)	https://github.com/apache/seatunnel/commit/f3499a6eeb	2.3.10
[Improve][dist]add shade check rule (#8136)	https://github.com/apache/seatunnel/commit/51ef800016	2.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
[Improve][Common] Introduce new error define rule (#5793)	https://github.com/apache/seatunnel/commit/9d1b2582b2	2.3.4
[Improve] Remove use `SeaTunnelSink::getConsumedType` method and mark it as deprecated (#5755)	https://github.com/apache/seatunnel/commit/8de7408100	2.3.4
Support config column/primaryKey/constraintKey in schema (#5564)	https://github.com/apache/seatunnel/commit/eac76b4e50	2.3.4
[Doc] update iotdb document (#5404)	https://github.com/apache/seatunnel/commit/856aedb3c9	2.3.4
[Improve][Connector-V2] Remove scheduler in IoTDB sink (#5270)	https://github.com/apache/seatunnel/commit/299637868c	2.3.4
[Hotfix] Fix com.google.common.base.Preconditions to seatunnel shade one (#5284)	https://github.com/apache/seatunnel/commit/ed5eadcf73	2.3.3
Merge branch 'dev' into merge/cdc	https://github.com/apache/seatunnel/commit/4324ee1912	2.3.1
[Improve][Project] Code format with spotless plugin.	https://github.com/apache/seatunnel/commit/423b583038	2.3.1
[improve][api] Refactoring schema parse (#4157)	https://github.com/apache/seatunnel/commit/b2f573a13e	2.3.1
[Improve][build] Give the maven module a human readable name (#4114)	https://github.com/apache/seatunnel/commit/d7cd601051	2.3.1
[Improve][Project] Code format with spotless plugin. (#4101)	https://github.com/apache/seatunnel/commit/a2ab166561	2.3.1
[Improve][SourceConnector] Unified schema parameter, update IoTDB sou… (#3896)	https://github.com/apache/seatunnel/commit/a0959c5fd1	2.3.1
[Feature][Connector] add get source method to all source connector (#3846)	https://github.com/apache/seatunnel/commit/417178fb84	2.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)	https://github.com/apache/seatunnel/commit/b9164b8ba1	2.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)	https://github.com/apache/seatunnel/commit/226dc6a119	2.3.0
[Improve][Connector-V2][Iotdb] Unified exception for iotdb source & sink connector (#3557)	https://github.com/apache/seatunnel/commit/7353fed6d6	2.3.0
[Feature][Connector V2] expose configurable options in IoTDB (#3387)	https://github.com/apache/seatunnel/commit/06359ea76a	2.3.0
[Improve][Connector-V2][IotDB]Add IotDB sink parameter check (#3412)	https://github.com/apache/seatunnel/commit/91240a3dcb	2.3.0
[Bug][Connector-v2] Fix IoTDB connector sink NPE (#3080)	https://github.com/apache/seatunnel/commit/e5edf02433	2.3.0-beta
[Imporve][Connector-V2] Imporve iotdb connector (#2917)	https://github.com/apache/seatunnel/commit/3da11ce19b	2.3.0-beta
[DEV][Api] Replace SeaTunnelContext with JobContext and remove singleton pattern (#2706)	https://github.com/apache/seatunnel/commit/cbf82f755c	2.2.0-beta
[#2606]Dependency management split (#2630)	https://github.com/apache/seatunnel/commit/fc047be69b	2.2.0-beta
[chore][connector-common] Rename SeatunnelSchema to SeaTunnelSchema (#2538)	https://github.com/apache/seatunnel/commit/7dc2a27388	2.2.0-beta
[Connectors-V2]Support IoTDB Source (#2431)	https://github.com/apache/seatunnel/commit/7b78d6c922	2.2.0-beta
[Feature][Connector-V2] Support IoTDB sink (#2407)	https://github.com/apache/seatunnel/commit/c1bbbd59d5	2.2.0-beta

IoTDBv2

Support Those Engines​

Description​

Key Features​

Supported DataSource Info​

Data Type Mapping​

Sink Options​

Examples​

Example 1: Write data to IoTDB-tree​

Case 1​

Case 2​

Case 3​

Example 2： Write data into IoTDB-table​

Case 1​

Case 2​

Case 3​

Changelog​

Support Those Engines

Description

Key Features

Supported DataSource Info

Data Type Mapping

Sink Options

Examples

Example 1: Write data to IoTDB-tree

Case 1

Case 2

Case 3

Example 2： Write data into IoTDB-table

Case 1

Case 2

Case 3

Changelog