近日，百度云天工物联网场景正式推出流式计算服务，填补了规则引擎、函数计算等只能针对单个数据流、单个消息处理的局限性。流式计算服务的推出，意味着用户在物联网场景中可利用流式计算服务对单个数据流的前后多个消息，或者多个数据流之间进行聚合计算。

通过流式计算的聚合运算能力，比如求平均、和、最大值、最小值等，可以关联分析多个流之间的联系等。接下来，让我们通过具体案例来了解下流式计算的作用。

以某烟草公司为例，其拥有1000多个烤烟房，每个烤烟房通过一个物解析网关，每分钟采集一次烤烟房的温度、湿度等数据，报送到云端，以实时监控烤烟房内部温湿度是否正常。通过为每一个烤烟房建立一条告警，设定越限条件，当该烤烟房的温湿度越限，则触发短信告警。

不过这种解决方案有个先决条件，即：网关总是能正常地上报数据。但是，导致网关无法上报数据的内外在因素存在很多，比如供电故障、设备故障、网络信号差、环境温度过高等等，这些因素都有可能导致网关意外停止工作。

设计流式计算监控。针对这种情况，流式计算就可以派上大用场，可以巧妙利用流式计算的窗口和时间概念，设计出流式监控，可以实现一个任务就监控上述1000多个设备，并且可以实时、准确报告具体哪个网关在什么时候停止了数据的上报;而增加网关设备时，无需修改任务。

具体解决办法。在物解析的轮询请求设置中，设置网关对Modbus设备的采集周期为60秒，并且每个解析项目解析后的数据转到统一的物接入主题：/modbus/parsed;随后在流式计算中创建名为:device_offine_monitor_src的数据源，数据源的物接入主题即来自上面解析后的主题：/modbus/parsed;流式计算任务中使用的时间类型为PROCESSTIME，可以实现即便监控一个设备，也照样工作;当流式计算任务中的会话窗口一直有数据流进来时，窗口始终不会关闭，不会触发告警;而当一旦某个设备在规定时间内没有数据流进来，则对应的会话窗口就会关闭，并输出一条信息，触发告警。

比如，会话窗口选择了65秒。理想情况下，只要60秒没有收到消，即认为网关故障，考虑到网络传输可能导致的延迟，增加了5秒的等待时间，以免因细小的传输延迟导致误报。

具体技术步骤：

1，在物解析的轮询请求设置中，设置网关对Modbus设备的采集周期为60秒

2，每个解析项目解析后的数据转到统一的物接入主题：/modbus/parsed

解析后的数据格式(关键部分)如下：

{

"gatewayid": "3faedb39-77ff-45b6-ac7e-c1739077e4d5"

}

因为只对网关id进行监控，其他数据无关，因此省略。

在流式计算中：

3，创建一个数据源，名为: device_offline_monitor_src;数据源的物接入主题来自上面解析后的主题：/modbus/parsed

4，创建MQTT数据目的地，名为: device_offline_notice_sink，以接收异常设备告警。物接入主题为:/modbus/device/offline_notice

5，创建流式计算任务

时间类型选：PROCESSTIME

SQL 语句为：

INSERT INTO device_offline_notice_sink

SELECT

Session_End(rowtime, INTERVAL "65" SECOND ) AS ts_end,

gatewayid

FROM device_offline_monitor_src

GROUP BY

Session(rowtime, INTERVAL "65" SECOND ),

gatewayid

当异常发生时，从数据目的地主题/modbus/device/offline_notice能收到如下格式的消息：

{

"ts_end": 1526459666537,

}

其中：ts_end为设备停止上报数据的时间，单位为毫秒。即设备上一次上报的时间为ts_end – 65000。gatewayid为网关id。

说明：

1，这里演示的是输出MQTT报警消息，事实上客户通过规则引擎，将该MQTT消息转化成短信告警。

2，流式计算任务里面使用的时间类型为PROCESSTIME，这样即便监控的是一个设备，也能照样工作。

3，流式计算任务里面采用的会话窗口，当一直有数据流进来，窗口始终不关闭，也就是始终没有输出。一旦某个设备规定时间没有数据流进来，对应的会话窗口就会关闭，就会输出一条消息。以此触发告警。

4，会话窗口选择了65秒。理想情况下，只要60秒没有收到消息，即认为网关故障。但考虑到网络传输上可能导致的延迟，因此，这里再增加了5秒等待，以免因细小的传输延迟导致误报。

现在，是不是比较了解流式计算了呢?

可以说，流式计算在物联网场景中有着广泛的应用场景，百度云天工的流式计算服务让百度云天工解决方案更加完善，通过与规则引擎、函数计算的配合，让用户在物联网场景中的对数据流的实时处理更加科学和准确。

特别声明：本文为网易自媒体平台“网易号”作者上传并发布，仅代表该作者观点。网易仅提供信息发布平台。