---
title: IoTDB 数据集成
---

# IoTDB 数据集成

本文介绍如何将 FlowMQ Stream 中的设备遥测数据实时写入 Apache IoTDB。

## 适用场景

- 工业设备监控
- 能源与环境采集
- 海量时序数据持久化

## 前提条件

- 已完成基础数据流准备，参考 [数据集成总览](data-integration.md)
- IoTDB 服务可访问
- 已具备写入目标路径的权限

## 配置步骤

1. 在“数据管道”模块点击“创建管道”，填写名称（如 `iotdb管道`），方向选择下行，任务数按需设置。

2. 选择数据流（如 `flowmq/mqtt/kafka`），点击“继续”。

3. 选择“iotdb”卡片，填写连接参数：

| 参数      | 值 |
|-----------|----|
| urls      | 192.168.55.31:6667 |
| username  | root |
| password  | root |
| device_id | `root.ln.${! metadata("kafka_topic").re_replace_all("/", ".") }` |

4. 配置 `columns` 字段映射：

| timestamp | measurement | data_type | value |
|-----------|-------------|-----------|-------|
| `${! json("data.timestamp") }` | `${! json("source") }` | FLOAT | `${! json("data.value") }` |

5. 点击“测试”验证目标连接。
![iotdb第三步](.assets/data-integration5_cn.png)

6. 点击“继续”，确认配置并创建 Pipeline。

7. 在管道列表点击“开始”运行。运行后可在 IoTDB 中查看写入结果（例如路径 `root.ln.flowmq.mqtt.kafka`）。

## 参数与表达式说明

- `device_id`：`root.ln.${! metadata("kafka_topic").re_replace_all("/", ".") }`
将主题中的 `/` 替换为 `.`，拼接成 IoTDB 设备路径。

- `timestamp`：`json("data.timestamp")`
提取时序时间戳。

- `measurement`：`json("source")`
提取测点名称（如传感器 ID）。

- `value`：`json("data.value")`
提取测点数值。

## 配置建议

- 统一规划 `device_id` 命名层级，避免路径混乱。
- 关注 `timestamp` 精度与 IoTDB 配置一致性。
- 对异常值可在上游或转换阶段做清洗后再写入。