DataX
大约 4 分钟
DataX
1. 概述
DataX 是阿里开源的离线数据同步框架，通过插件机制实现异构数据源高效批量读写；IoTDB 提供 IoTDBReader 与 IoTDBWriter 两款插件，完成与 DataX 的无缝集成，从而支持把时序数据从 IoTDB 高速导出到任意目标端，也能把外部数据一键写入 IoTDB，实现 IoTDB 与全域数据湖仓的灵活互通。
2. 使用步骤
2.1 环境要求
| 软件名称 | 版本要求 | 安装说明 |
|---|---|---|
| IoTDB | >=2.0.5.1 | 快速上手 |
| DataX | 定制版 | 联系天谋工作人员获取安装包 |
2.2 读取数据 IoTDBReader
IoTDBReader 支持从 IoTDB 数据库中读取数据。
2.2.1 参数说明
| 参数名 | 描述 | 是否必选 | 默认值 |
|---|---|---|---|
| username | 用户名 | 是 | 无 |
| password | 用户名的密码 | 是 | 无 |
| host | 连接 IoTDB 数据库的主机地址 | 是 | 无 |
| port | 端口 | 是 | 无 |
| version | IoTDB 版本 | 是 | 无 |
| timeColumnPosition | 时间列在 Record 中列的位置 | 否 | 0 |
| querySqls | 直接写多行 SQL，可以并行读取，此时下面的参数失效。 | 否 | 无 |
| device | IoTDB 中的概念，可理解为 mysql 中的表。 | querySqls 为空时必选 | 无 |
| measurements | IoTDB 中的概念，可理解为 mysql 中的字段。 | querySqls 为空时必选 | 无 |
| where | 查询条件 | 否 | 无 |
2.2.2 配置样例
配置一个从 IoTDB 抽取数据并写入 mysql 的作业
- IoTDB 中的源数据如下
IoTDB> show timeseries root.mysql.**
+-----------------------------+-----+----------+--------+--------+-----------+----+----------+--------+------------------+--------+
| Timeseries|Alias| Database|DataType|Encoding|Compression|Tags|Attributes|Deadband|DeadbandParameters|ViewType|
+-----------------------------+-----+----------+--------+--------+-----------+----+----------+--------+------------------+--------+
| root.mysql.device.city| null|root.mysql| TEXT| PLAIN| LZ4|null| null| null| null| BASE|
|root.mysql.device.temperature| null|root.mysql| DOUBLE| GORILLA| LZ4|null| null| null| null| BASE|
| root.mysql.device.humidity| null|root.mysql| DOUBLE| GORILLA| LZ4|null| null| null| null| BASE|
| root.mysql.device.status| null|root.mysql| BOOLEAN| RLE| LZ4|null| null| null| null| BASE|
+-----------------------------+-----+----------+--------+--------+-----------+----+----------+--------+------------------+--------+
IoTDB> select * from root.mysql.** align by device
+-----------------------------+-----------------+----+-----------+--------+------+
| Time| Device|city|temperature|humidity|status|
+-----------------------------+-----------------+----+-----------+--------+------+
|2025-08-19T08:00:00.000+08:00|root.mysql.device| bj| 31.45| 66.08| true|
|2025-08-19T09:00:00.000+08:00|root.mysql.device| bj| 20.86| 60.32| false|
|2025-08-19T11:00:00.000+08:00|root.mysql.device| bj| 28.73| 63.21| false|
|2025-08-19T12:00:00.000+08:00|root.mysql.device| bj| 21.82| 48.58| true|
+-----------------------------+-----------------+----+-----------+--------+------+- iotdb_to_mysql_job.json 文件配置如下：
{
"job": {
"setting": {
"speed": {
"channel": 3
}
},
"content": [
{
"reader": {
"name": "iotdbreader",
"parameter": {
"username": "root",
"password": "root",
"host": "127.0.0.1",
"port": 6668,
"fetchSize": 10000,
"version": "V_1_0",
"##": "时间列插入DataX的Record中的位置，默认第0列",
"timeColumnPosition": 0,
"##":"写了querySqls 默认下面全部参数失效，适合开发人员, 多个session并行执行多条SQL语句",
"querySqls":[
],
"device": "root.mysql.device",
"##":"时间列不属于测点",
"measurements": ["city","temperature","humidity","status"],
"where": "time > 2025-08-19 00:00:00 and time < 2025-08-19 20:00:00"
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {
"username": "root",
"password": "IoTDB@2024",
"writeMode": "insert",
"## 需要提前建表": "CREATE TABLE device (`time` DATETIME,`city` VARCHAR(255),`temperature` DOUBLE,`humidity` DOUBLE,`status` BOOLEAN);",
"column": ["time","city","temperature","humidity","status"],
"session": [
"set session sql_mode='ANSI'"
],
"preSql": [
"delete from device"
],
"connection": [
{
"table": [
"device"
], "##": "下面的URL需要把中括号去掉，否则报错，mysqlreader的bug，未修改",
"jdbcUrl": "jdbc:mysql://localhost:3306/demo?useUnicode=true&allowPublicKeyRetrieval=true&characterEncoding=utf-8"
}
]
}
}
}
]
}
}提前在 mysql 中创建表 device
执行 dataX 程序：
python bin/datax.py job/iotdb_to_mysql_job.json- 验证：在 mysql 中查询到数据
mysql> select * from device;
+---------------------+------+-------------+----------+--------+
| time | city | temperature | humidity | status |
+---------------------+------+-------------+----------+--------+
| 2025-08-19 08:00:00 | bj | 31.45 | 66.08 | 1 |
| 2025-08-19 09:00:00 | bj | 20.86 | 60.32 | 0 |
| 2025-08-19 11:00:00 | bj | 28.73 | 63.21 | 0 |
| 2025-08-19 12:00:00 | bj | 21.82 | 48.58 | 1 |
+---------------------+------+-------------+----------+--------+2.3 写入数据 IoTDBWriter
IoTDBWriter 支持写入数据到 IoTDB 数据库目标表(设备)。
2.3.1 参数说明
| 参数名 | 描述 | 是否必选 | 默认值 |
|---|---|---|---|
| username | 用户名 | 是 | 无 |
| password | 用户名的密码 | 是 | 无 |
| host | 连接 IoTDB 数据库的主机地址 | 是 | 无 |
| port | 端口 | 是 | 无 |
| version | IoTDB 版本 | 是 | 无 |
| timeColumnPosition | 时间列在 Record 中列的位置 | 否 | 0 |
| device | IoTDB 中的概念，对应 mysql 中的表名。 | 是 | 无 |
| measurements | IoTDB 中的概念，对应 mysql 中的字段集合，顺序应该与 record 中 column 的顺序相同。 | 是 | 无 |
| batchSize | 每 batchSize 条 record 为一个 batch 进行写入 | 否 | 1000 |
| preSql | 插入前是否预先执行 SQL | 否 | 无 |
2.3.2 配置样例
配置一个 MySQL 数据写入 IoTDB 的作业 ：
- mysql 中的数据如下
mysql> select * from device;
+---------------------+------+-------------+----------+--------+
| time | city | temperature | humidity | status |
+---------------------+------+-------------+----------+--------+
| 2025-08-19 08:00:00 | bj | 31.45 | 66.08 | 1 |
| 2025-08-19 09:00:00 | bj | 20.86 | 60.32 | 0 |
| 2025-08-19 11:00:00 | bj | 28.73 | 63.21 | 0 |
| 2025-08-19 12:00:00 | bj | 21.82 | 48.58 | 1 |
+---------------------+------+-------------+----------+--------+- iotdb_from_mysql_job.json 文件配置如下
{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"username": "root",
"password": "IoTDB@2024",
"column": ["time","city","temperature","humidity","status"],
"splitPk": "",
"connection": [
{
"table": [
"device"
],
"jdbcUrl": [
"jdbc:mysql://localhost:3306/demo?useUnicode=true&allowPublicKeyRetrieval=true&characterEncoding=utf-8"
]
}
]
}
},
"writer": {
"name": "iotdbwriter",
"parameter": {
"username": "root",
"password": "root",
"host": "127.0.0.1",
"port": 6668,
"version": "V_1_0",
"##": "Reader中时间列的位置，默认0列",
"timeColumnPosition": 0,
"batchSize": 1000,
"device": "root.db.device",
"measurements": ["city","temperature","humidity","status"],
"###": "在插入之前，预先执行的SQL，默认为空",
"preSql": []
}
}
}
],
"setting": {
"speed": {
"channel": 3
}
}
}
}- 执行 dataX 程序
python bin/datax.py job/iotdb_from_mysql_job.json- 验证：在iotdb中查询到数据
IoTDB> select * from root.db.** align by device
+-----------------------------+--------------+----+-----------+--------+------+
| Time| Device|city|temperature|humidity|status|
+-----------------------------+--------------+----+-----------+--------+------+
|2025-08-19T08:00:00.000+08:00|root.db.device| bj| 31.45| 66.08| true|
|2025-08-19T09:00:00.000+08:00|root.db.device| bj| 20.86| 60.32| false|
|2025-08-19T11:00:00.000+08:00|root.db.device| bj| 28.73| 63.21| false|
|2025-08-19T12:00:00.000+08:00|root.db.device| bj| 21.82| 48.58| true|
+-----------------------------+--------------+----+-----------+--------+------+