Version: Next

IoTDB

IoTDB sink connector

Support Those Engines

Spark
Flink
SeaTunnel Zeta

Description

Used to write data to IoTDB.

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.
cdc

The IoTDB sink connector writes rows by calling the IoTDB insert RPC. When a row carries a non-unique (device, timestamp) pair, the write is treated as an upsert — the latest value overwrites earlier ones — so duplicate deliveries from upstream do not create phantom rows. Row-kind UPDATE/DELETE are not interpreted as CDC operations; all rows are written as inserts.

Supported DataSource Info

Datasource	Supported Versions	Url
IoTDB	`0.13.0 <= version <= 1.3.X`	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

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 user username
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 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	String	No	-	Specify device storage group(path prefix) example: deviceId = \${storage_group} + "." + \${key_device}
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`
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
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

Write Rules

key_device must name the SeaTunnel field that contains the IoTDB device path.
storage_group is a string prefix. When it is set, the final device path is built from storage_group and the value of key_device.
key_timestamp can name a STRING, BIGINT, or TIMESTAMP field. If it is not configured, the connector uses the current processing time.
If key_measurement_fields is not configured, all fields except key_device and key_timestamp are written as measurements.
The sink supports STRING, BOOLEAN, TINYINT, SMALLINT, INT, BIGINT, FLOAT, and DOUBLE measurement fields.

Examples

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

Case1

Only required options used:

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

sink {
  IoTDB {
    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|
+------------------------+------------------------+--------------+-----------+--------------+---------+---------+-----------+-----------+------+-----------+--------+---------+

Case2

Use source event's time:

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

sink {
  IoTDB {
    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|
+------------------------+------------------------+--------------+-----------+--------------+---------+---------+-----------+-----------+------+-----------+--------+---------+

Case3

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 {
  IoTDB {
    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|
+------------------------+------------------------+--------------+-----------+

Case4: Streaming writes with explicit batch flush

For long-running streaming jobs, increase batch_size to reduce per-row RPC overhead. The connector flushes the buffered rows when either the buffer fills up to batch_size or the checkpoint completes. Set max_retries and max_retry_backoff_ms to keep the job resilient against transient RPC failures.

env {
  parallelism = 2
  job.mode = "STREAMING"
  checkpoint.interval = 10000
}

sink {
  IoTDB {
    node_urls = ["localhost:6667", "localhost:6668"]
    username = "root"
    password = "root"
    key_device = "device_name"
    key_timestamp = "event_ts"
    batch_size = 2048
    max_retries = 3
    retry_backoff_multiplier_ms = 100
    max_retry_backoff_ms = 5000
  }
}

node_urls accepts multiple IoTDB nodes. The sink will pick one as the active write node per task and fall over to the others when the active node fails.

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

IoTDB

Support Those Engines​

Description​

Key Features​

Supported DataSource Info​

Data Type Mapping​

Sink Options​

Write Rules​

Examples​

Case1​

Case2​

Case3​

Case4: Streaming writes with explicit batch flush​

Changelog​

Support Those Engines

Description

Key Features

Supported DataSource Info

Data Type Mapping

Sink Options

Write Rules

Examples

Case1

Case2

Case3

Case4: Streaming writes with explicit batch flush

Changelog