物联网场景中的实时计算问题与方案

Posted 2021-08-18 咬定青松

物联网应用中一种常见场景是：假设有一流量表（比如供气站的接纳量、风力发电站的风量流量计、居民用户表等），每隔1秒钟上报当前累积量，要求准实时统计该表具当日以及当月和当年的数据量。

分析：初看起来，这是小学生都会的数学减法问题：用当前值减去初始值。然而小学生的思维过于简单和理想化，现实很复杂，因为在自然环境中存放的表具会断电、表具会损坏、异常环境会让表具读数异常（如跳表）、表读数达到最大值等。如果用当前值减去初始值，会因为忽略这期间的异常变化而导致数据丢失，进而得到不正确的结果。解决该问题可以借鉴银行的存储余额的做法，除了有一个汇总余额的结果，还必须有一个”消费明细“用于对账和审计。在我们这个示例场景中，每秒钟1次上报就是明细数据，有了明细数据后就可以计算一段时间范围内的累计值，然后再把所有小范围内的累计值再次汇总，即最终的结果。但是小范围的时间范窗口取多大合适呢？如果时间窗口过小，可能会因为数据过小导致累计误差放大。时间窗口过大，无法满足业务需求，业务场景中一般会要求最近一分钟、一小时或者一天，都有可能，这取决于应用需求，假如应用的最小统计周期为一小时，那么统计窗口为1小时，而实际中一个完整的业务日往往也是精确到小时级别，比如今日的8点钟到明日的8点钟视为今日，此时按照一小时汇总是比较合适的值。接下来我们看如何做？

按照表具数量分为3种情况来分析：One、Many、too Many。

One

单表情况主要用于描述思路和演示，如下图所示：低频数据在高频数据基础上汇总，每个低频数据记录行格式为（统计周期、当前统计周期的初始值、当前周期内的累计值、日累计值、月累计值、年累计值），统计周期为唯一键，记录的数据更新由最高频数据到达触发，比如当原始秒级的时序数据到达，触发小时级及以上周期的行汇总数据同时”原地“更新：在同一数据行上更新累计值。其中当前小时统计周期内的累计值(acc_value_hour)为表具当前值与当前小时周期内的初始值的差值，其他周期内的累计值为当前小时统计周期内的累计值与上一小时统计周期内的累计值之和，比如图中红色箭头的累加方向为日汇总过程。

表具数据上报过程中，正常情况只增不减，但会存在异常值，异常值可能过大或者过小两种情况，过小比如下图中的示例：12点范围的初始值正常为93.5，但是到第3条记录出现时，该值远小于前面的记录，假设前面的记录正常，而该记录远小于前面记录的平均值，因此可以有理由认为表具出现变更重新计数，因此简单的做法可以将前面正常的累计值取负作为新的初始值。

而过大的情况比如下图中的示例：13点范围内的某条记录值远大于之前记录的平均值，此时有理由认为该值出现异常，对此做法可以根据情况自由而定，比如可以忽略该值，或者根据历史平均值对其修正。

Many

从单表的数据处理过程可知，小时周期数据汇总只跟当前记录行数据有关系，其他周期汇总数据跟前一条数据有关系，而在数据异常处理过程中，最多跟当前小时周期的数据量有关，而异常情况发生频率较低，因此通常情况下，每次更新记录涉及至少2次读写，该统计操作的复杂度为O(1)。对于有Many表的情况下，复杂度为O(n)，n为设备数量，基于OLTP处理模式处理，上述数据存储格式可增加设备标识：

此时常规的在线处理架构（如分布式服务和高并发读写系统）能胜任大部分以上场景的统计需求，然而其瓶颈是存储系统的读写效率和存储规模，比如假设场景有1000块不同的表具，数据上报频率为1次每秒，那么数据更新，读写操作频率为至少每秒2000次，每天的原始时序数据记录条数为1000*24*3600=8,6400,000，统计记录条数为1000*24条，如果提高表的数量或者数据上报频率，这个数据会更大。当然该场景还有很大优化空间，比如将原始时序表用其他系统存储，比如消息系统，数据统计结果更新也可以先合并后批量更新等。实际上这些优化方式已经进入OLAP模式了，或者说进入两者的中间地带。

too Many

分析上述OLTP模式，其瓶颈在于有速度无吞吐量，这也是批处理和”单个“处理的区别，那如何用批处理重做一遍呢？实际上根据日期维度，通过几条group by即可，group by的汇总数据可以实现分组内的自定义逻辑如异常值剔除等数据清洗、求平均值等。接下来需要重点考虑的是如何优化巨量时序数据存储的问题，常规方案很容易做到，比如这样：

使用Flink的时间范围为1小时的滚动窗口实现数据的聚合处理，然后将结果存储到数据库。既然说到这里，我们将问题描述得更为真实一些：比如某天然气加气站，有多个入口和多个出口，每个出入口都有多种类型的表具（计量表、温度表、压力表、阀门开关表等），如何实时计算并展示每个表具或站点的历史数据变化曲线？

分析：先看看上述架构能否满足，首先基于1小时的时间窗口计算，也能计算每小时内的累计流量以及最新瞬时流量、温度和压力值，然而如果在这个工艺图中实时监控各阀门开闭状态，如果不增加现有架构的复杂度情况下，能否实现呢？如果业务方要求缩小时间范围，每隔5分钟采集一个数据点，不增加现有架构的复杂度情况下，能否实现呢？留给读者思考吧，欢迎后台留言和讨论。