近日，百度云天工物聯(lián)網(wǎng)場景正式推出流式計(jì)算服務(wù)，填補(bǔ)了規(guī)則引擎、函數(shù)計(jì)算等只能針對單個(gè)數(shù)據(jù)流、單個(gè)消息處理的局限性。流式計(jì)算服務(wù)的推出，意味著用戶在物聯(lián)網(wǎng)場景中可利用流式計(jì)算服務(wù)對單個(gè)數(shù)據(jù)流的前后多個(gè)消息，或者多個(gè)數(shù)據(jù)流之間進(jìn)行聚合計(jì)算。

　　通過流式計(jì)算的聚合運(yùn)算能力，比如求平均、和、最大值、最小值等，可以關(guān)聯(lián)分析多個(gè)流之間的聯(lián)系等。接下來，讓我們通過具體案例來了解下流式計(jì)算的作用。

　　以某煙草公司為例，其擁有1000多個(gè)烤煙房，每個(gè)烤煙房通過一個(gè)物解析網(wǎng)關(guān)，每分鐘采集一次烤煙房的溫度、濕度等數(shù)據(jù)，報(bào)送到云端，以實(shí)時(shí)監(jiān)控烤煙房內(nèi)部溫濕度是否正常。通過為每一個(gè)烤煙房建立一條告警，設(shè)定越限條件，當(dāng)該烤煙房的溫濕度越限，則觸發(fā)短信告警。

　　不過這種解決方案有個(gè)先決條件，即：網(wǎng)關(guān)總是能正常地上報(bào)數(shù)據(jù)。但是，導(dǎo)致網(wǎng)關(guān)無法上報(bào)數(shù)據(jù)的內(nèi)外在因素存在很多，比如供電故障、設(shè)備故障、網(wǎng)絡(luò)信號(hào)差、環(huán)境溫度過高等等，這些因素都有可能導(dǎo)致網(wǎng)關(guān)意外停止工作。

　　設(shè)計(jì)流式計(jì)算監(jiān)控。針對這種情況，流式計(jì)算就可以派上大用場，可以巧妙利用流式計(jì)算的窗口和時(shí)間概念，設(shè)計(jì)出流式監(jiān)控，可以實(shí)現(xiàn)一個(gè)任務(wù)就監(jiān)控上述1000多個(gè)設(shè)備，并且可以實(shí)時(shí)、準(zhǔn)確報(bào)告具體哪個(gè)網(wǎng)關(guān)在什么時(shí)候停止了數(shù)據(jù)的上報(bào);而增加網(wǎng)關(guān)設(shè)備時(shí)，無需修改任務(wù)。

　　具體解決辦法。在物解析的輪詢請求設(shè)置中，設(shè)置網(wǎng)關(guān)對Modbus設(shè)備的采集周期為60秒，并且每個(gè)解析項(xiàng)目解析后的數(shù)據(jù)轉(zhuǎn)到統(tǒng)一的物接入主題：/modbus/parsed;隨后在流式計(jì)算中創(chuàng)建名為:device_offine_monitor_src的數(shù)據(jù)源，數(shù)據(jù)源的物接入主題即來自上面解析后的主題：/modbus/parsed;流式計(jì)算任務(wù)中使用的時(shí)間類型為PROCESSTIME，可以實(shí)現(xiàn)即便監(jiān)控一個(gè)設(shè)備，也照樣工作;當(dāng)流式計(jì)算任務(wù)中的會(huì)話窗口一直有數(shù)據(jù)流進(jìn)來時(shí)，窗口始終不會(huì)關(guān)閉，不會(huì)觸發(fā)告警;而當(dāng)一旦某個(gè)設(shè)備在規(guī)定時(shí)間內(nèi)沒有數(shù)據(jù)流進(jìn)來，則對應(yīng)的會(huì)話窗口就會(huì)關(guān)閉，并輸出一條信息，觸發(fā)告警。

　　比如，會(huì)話窗口選擇了65秒。理想情況下，只要60秒沒有收到消，即認(rèn)為網(wǎng)關(guān)故障，考慮到網(wǎng)絡(luò)傳輸可能導(dǎo)致的延遲，增加了5秒的等待時(shí)間，以免因細(xì)小的傳輸延遲導(dǎo)致誤報(bào)。

　　具體技術(shù)步驟：

　　1，在物解析的輪詢請求設(shè)置中，設(shè)置網(wǎng)關(guān)對Modbus設(shè)備的采集周期為60秒

　　2，每個(gè)解析項(xiàng)目解析后的數(shù)據(jù)轉(zhuǎn)到統(tǒng)一的物接入主題：/modbus/parsed

　　解析后的數(shù)據(jù)格式(關(guān)鍵部分)如下：

　　{

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

　　}

　　因?yàn)橹粚�網(wǎng)關(guān)id進(jìn)行監(jiān)控，其他數(shù)據(jù)無關(guān)，因此省略。

　　在流式計(jì)算中：

　　3，創(chuàng)建一個(gè)數(shù)據(jù)源，名為: device_offline_monitor_src;數(shù)據(jù)源的物接入主題來自上面解析后的主題：/modbus/parsed

　　4，創(chuàng)建MQTT數(shù)據(jù)目的地，名為: device_offline_notice_sink，以接收異常設(shè)備告警。物接入主題為:/modbus/device/offline_notice

　　5，創(chuàng)建流式計(jì)算任務(wù)

　　時(shí)間類型選：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

　　當(dāng)異常發(fā)生時(shí)，從數(shù)據(jù)目的地主題/modbus/device/offline_notice能收到如下格式的消息：

　　{

　　"ts_end": 1526459666537,

　　}

　　其中：ts_end為設(shè)備停止上報(bào)數(shù)據(jù)的時(shí)間，單位為毫秒。即設(shè)備上一次上報(bào)的時(shí)間為ts_end – 65000。gatewayid為網(wǎng)關(guān)id。

　　說明：

　　1，這里演示的是輸出MQTT報(bào)警消息，事實(shí)上客戶通過規(guī)則引擎，將該MQTT消息轉(zhuǎn)化成短信告警。

　　2，流式計(jì)算任務(wù)里面使用的時(shí)間類型為PROCESSTIME，這樣即便監(jiān)控的是一個(gè)設(shè)備，也能照樣工作。

　　3，流式計(jì)算任務(wù)里面采用的會(huì)話窗口，當(dāng)一直有數(shù)據(jù)流進(jìn)來，窗口始終不關(guān)閉，也就是始終沒有輸出。一旦某個(gè)設(shè)備規(guī)定時(shí)間沒有數(shù)據(jù)流進(jìn)來，對應(yīng)的會(huì)話窗口就會(huì)關(guān)閉，就會(huì)輸出一條消息。以此觸發(fā)告警。

　　4，會(huì)話窗口選擇了65秒。理想情況下，只要60秒沒有收到消息，即認(rèn)為網(wǎng)關(guān)故障。但考慮到網(wǎng)絡(luò)傳輸上可能導(dǎo)致的延遲，因此，這里再增加了5秒等待，以免因細(xì)小的傳輸延遲導(dǎo)致誤報(bào)。

　　現(xiàn)在，是不是比較了解流式計(jì)算了呢?

　　可以說，流式計(jì)算在物聯(lián)網(wǎng)場景中有著廣泛的應(yīng)用場景，百度云天工的流式計(jì)算服務(wù)讓百度云天工解決方案更加完善，通過與規(guī)則引擎、函數(shù)計(jì)算的配合，讓用戶在物聯(lián)網(wǎng)場景中的對數(shù)據(jù)流的實(shí)時(shí)處理更加科學(xué)和準(zhǔn)確。

　　特別聲明：本文為網(wǎng)易自媒體平臺(tái)“網(wǎng)易號(hào)”作者上傳并發(fā)布，僅代表該作者觀點(diǎn)。網(wǎng)易僅提供信息發(fā)布平臺(tái)。