4、遥测数据集成：遥测数据上报、批量数据上报、数据查询与聚合

遥测数据，说白了就是设备上报的那些「心跳」和「状态」。温度、湿度、电压、转速……这些数据才是物联网系统的真正血液。我见过不少项目，设备连上了，数据也收了，但到了分析阶段才发现——数据格式不对、时间戳乱跳、批量上报时把系统搞崩了。

嗯，今天我们就来把这几个硬骨头啃下来。

4.1 单点遥测上报：最基础的姿势

先看最简单的场景：一个传感器，每隔几秒上报一个温度值。在 ThingsBoard 里，这通常通过 MQTT 的 v1/devices/me/telemetry 主题来实现。

举个例子，一个温度传感器上报数据：

// 设备端 MQTT 发布
Topic: v1/devices/me/telemetry
Payload: {"temperature": 25.6, "humidity": 68.2}

我在项目中遇到过一个问题：设备上报的 key 带空格，比如 "temp value"。ThingsBoard 虽然能接收，但后续做规则链处理时，变量引用会非常麻烦。我的建议是——key 命名统一用下划线，别给自己挖坑。

4.2 批量数据上报：别一条一条发

你想想看，如果一个设备有 10 个传感器，每个传感器每秒上报一次，那每秒就是 10 条 MQTT 消息。网络开销大，服务端压力也大。

更好的做法是批量上报。ThingsBoard 支持在同一个 Payload 里放多个数据点：

// 批量上报 - 方式一：多个 key 同时上报
{
  "temperature": 25.6,
  "humidity": 68.2,
  "pressure": 1013.2
}

// 批量上报 - 方式二：同一个 key 的多个时间点数据
{
  "ts": [1620000000000, 1620000010000, 1620000020000],
  "values": [25.6, 25.8, 26.0]
}

方式一适合「多个指标同时采集」的场景。方式二适合「历史数据补录」——比如设备离线了一段时间，重新上线后把缓存的数据一次性补上来。

4.3 数据查询：REST API 的正确打开方式

数据存进去了，怎么查？ThingsBoard 提供了丰富的 REST API 来查询遥测数据。

最常用的查询接口：

// 查询设备最新遥测数据
GET /api/plugins/telemetry/DEVICE/{deviceId}/values/timeseries?keys=temperature,humidity

// 查询历史遥测数据（带时间范围）
GET /api/plugins/telemetry/DEVICE/{deviceId}/values/timeseries?keys=temperature&startTs=1620000000000&endTs=1620100000000&limit=100

这里有个坑要注意——startTs 和 endTs 的单位是毫秒。我见过有人传了秒级时间戳，结果查出来全是空数据。排查了半天才发现是单位搞错了。

另外，limit 参数默认是 100，最大可以设到 1000。如果你需要查更多数据，建议分页查询，或者用后面的聚合查询。

4.4 数据聚合：从海量数据中提炼价值

原始数据太多了，每秒一条，一天就是 86400 条。做报表时根本用不上这么细的粒度。这时候就需要聚合查询。

ThingsBoard 支持按时间窗口做聚合：

// 按小时聚合，查询平均值
GET /api/plugins/telemetry/DEVICE/{deviceId}/values/timeseries?keys=temperature&
     startTs=1620000000000&endTs=1620100000000&
     interval=3600000&agg=AVG

// 支持的聚合函数：AVG, MIN, MAX, SUM, COUNT, NONE

interval 的单位也是毫秒。3600000 毫秒 = 1 小时。返回的结果会按小时分组，每组一个聚合值。

聚合函数	说明	适用场景
AVG	平均值	温度、湿度等平稳指标
MIN	最小值	监控最低电压、最低水位
MAX	最大值	监控峰值电流、最高温度
SUM	总和	累计用电量、总流量
COUNT	数据点数量	统计上报频率是否正常

4.5 实战：用规则链处理遥测数据

光会 API 调用还不够。在实际项目中，我更推荐用规则链来做数据预处理。比如：

• 过滤掉异常值（温度超过 100°C 直接丢弃）
• 数据格式转换（把华氏度转成摄氏度）
• 触发告警（温度超过阈值时发邮件）

一个典型的规则链配置：

// 规则链节点配置（简化版）
1. 消息类型节点：筛选遥测数据
2. 脚本节点：检查温度是否在合理范围
   - 脚本：return msg.temperature > -50 && msg.temperature < 100;
3. 转换节点：添加处理时间戳
   - 脚本：msg.processed_at = new Date().getTime();
4. 保存节点：存入数据库
5. 告警节点：如果温度 > 80，触发告警

我曾经在一个冷链物流项目里，用规则链做了「温度漂移检测」。当连续 3 个数据点的温度变化超过 5°C 时，自动触发告警。这个逻辑用代码写可能要几十行，但在规则链里拖拽几个节点就搞定了。

4.6 避坑指南

最后分享几个我踩过的坑：

• 时间戳问题：设备端的时间可能不准。我建议在规则链里用 metadata.ts 替换设备上报的时间戳，以服务器时间为准。
• 数据乱序：网络延迟可能导致数据到达顺序和采集顺序不一致。批量上报时，建议每条数据都带独立的时间戳。
• Key 数量限制：单个设备最多支持 1000 个不同的遥测 key。别把每个传感器的每个指标都设成独立 key，可以用 JSON 嵌套来减少 key 的数量。

嗯，遥测数据集成这块，说白了就是「收得进来、存得下来、查得出去」。把这三点做好了，剩下的就是业务逻辑的事了。下一章我们会聊告警与通知，到时候再细说怎么让系统自己「发现问题」。