數據異常率預警在工業安全生產中是一項重要工作，對於監控生產過程的穩定性，保障生產數據的有效性，維護生產設備的可靠性具有重要意義。隨着大數據技術在生產領域的深入應用，基於機器學習的智能預警已經成為各大生產企業進行生產數據異常率預警的一種有效解決方案。

1. 概要

本教程基於一個典型的物聯網企業生產場景，利用 DolphinDB 內置的機器學習框架中的 KNN（K-Nearest Neighbor，最經典和最簡單的有監督學習方法之一）算法構建迴歸模型，對實時數據異常率預警過程進行模擬。

2. 機器學習在物聯網中的應用場景

2.1 工業物聯網智能運維

隨着整個物聯網系統數據規模的急劇膨脹，以及服務類型的日趨多樣化、複雜化，基於人為指定規則的手工運維和在此基礎上加入自動化腳本實現的自動化運維已無法滿足大規模的運維需求。

為解決人為指定規則帶來的弊端，隨着智能時代的到來，智能運維的概念逐漸浮出水面。與人為指定規則不同的是，智能運維強調利用機器學習的算法從海量運維數據中總結規則，打破了人為指定規則的侷限性。

簡而言之，智能運維在自動化運維的基礎上加入了機器學習分析數據的環節，監測系統採集運維日誌數據，機器學習根據日誌數據做出分析並生成決策，最後指揮自動化腳本去執行決策，從而達到運維繫統的整體目標。

2.2 工業設備使用壽命預測

工業設備的損耗和故障在長期運作過程中是難以避免的，如果不能有效監控設備損耗情況的數據變化和應對可能發生的故障，將造成生產中斷、經濟損失，甚至危害公眾安全，造成人員傷亡。如果能對設備故障進行預警並提前制定應對方案，將保障設備可靠性和安全性，提高生產效率，實現利益最大化。

設備剩餘使用壽命預測旨在根據設備的運行狀態數據，結合設備的退化機理，利用人工智能技術對設備未來的故障發生時段進行預測。若能提前預測出設備的剩餘使用壽命，在設備故障發生前進行預警，就能幫助維護人員及早制定出最優應對方案。

2.3 工業生產異常數據預警

異常數據預警在工業生產中是一個相當重要的環節，已有越來越多的企業將其納入生產安全保障過程中的一個必要環節。其對於監控設備的生產環境，維護生產數據的價值性，提高生產效率具有重要意義。

異常數據預警具體實現：通過採集歷史數據，利用自定算法或機器學習算法分析歷史數據特徵並預測未來數據，最後根據特定的預警規則對未來數據進行預警判斷。

3.KNN

3.1 KNN 介紹

KNN ( K-NearestNeighbor ) 算法又叫 K 鄰近算法，是一種比較簡單的機器學習算法。該算法是一種監督學習算法，即可以用於分類任務，也可以用於迴歸任務。

• 監督學習算法：算法的輸入數據帶有標籤。

• 分類任務：算法的結果為一個決策面，輸出數據為離散數據。

• 迴歸任務：算法的結果為一條擬合線，輸出數據為連續數據。

KNN 分類算法思路：設置近鄰個數 K，如果一個樣本在特徵空間中的 K 個鄰近樣本中的大多數屬於某一個類別，則該樣本也屬於這個類別。

KNN 迴歸算法思路：設置近鄰個數 K，計算一個樣本在特徵空間中的 K 個近鄰樣本的平均值，作為該樣本的預測值。

3.2 KNN 優勢

• 模型簡單，無需過多調參就可以得到一個不錯的性能。

• 精度高。

• 對異常值不敏感。

• 無數據輸入假定。

3.3 K 值

3.3.1 評估指標

對於迴歸任務，這裏選擇一個常用的迴歸評估指標：根均方誤差（Root Mean Square Error, RMSE）。

RMSE 指標可以評價數據的變化程度，RMSE 越小，説明模型的擬合實驗數據能力越強。

3.3.2 K 值的選擇

K 值選擇過小，容易造成模型過擬合；K值選擇過大，容易造成模型欠擬合。

RMSE 越小，模型的擬合效果越好。本場景經過多次實驗，最終實驗的 K 值選擇為200。

4. 解決方案

DolphinDB 是由浙江智臾科技有限公司研發的一款高性能分佈式時序數據庫，集成了功能強大的編程語言和高容量高速度的流數據分析系統。針對用户對工業生產中異常數據預警的需求，本文提供了一個完整的解決方案，幫助用户掌握如何使用 DolphinDB 預警異常數據，以便更好地保障工業生產安全。

4.1 場景描述

生產場景為一個大型風電廠的發電量監控場景，風機監控人員需要實時監控發電量的平穩性，當檢測到風機發電量出現異常時，監控人員需要遠程實時調整風機參數以維持發電量的平穩。

為提前感知發電異常及時調整風機參數，風機工廠現決定利用影響風機發電的主要因素的監測數據，對未來一段時間的風機發電量進行預測，並將未來發電量與當前發電量進行異常比對。當異常率超過一定閾值時向風機監控人員遠程報警。為了便於理解 DolphinDB 如何實現機器學習，本章簡化了複雜的真實場景，只使用5個指標：風速、濕度、氣壓、温度和設備使用壽命，來進行指代。同時，由於生產數據的敏感性，進行了 DolphinDB 數據仿真操作，使用模擬數據代替真實記錄集。

• 設備數目：100台

• 時間單位：毫秒級 （1 ms = 0.001 s）

• 單台設備採集數據傳輸量：1條/ms

• 總採集數據傳輸量：10萬條/s

數據形式：

4.2 架構方案

由於真實生產場景的私密性和真實生產數據的敏感性，以及為了方便理解，除實驗設置的數據生產場景有所簡化外，其餘場景都將與真實生產場景保持一致。

當生產任務開啟後，模型預測任務與預警計算任務將同步開啟。運維人員可登錄Grafana，實時監控異常情況，並根據預警結果實時調整風機設備參數。

4.2.1 模型預測場景

模型預測場景流程設計如下：

1. 每10秒鐘用前10秒採集到的100萬條數據，訓練出一個 KNN 迴歸模型。

1. 用前10秒的100萬條數據，進行聚合計算，將10秒產生的100萬條數據聚合成1秒產生的10萬條數據。

1. 用 KNN 模型 對聚合數據（10萬條）進行預測，預測未來第10s的數據（10萬條）。

1. 將未來第10s的數據（10萬條）與10s聚合的數據（10萬條）輸入預警模塊進行預警計算。

4.2.2 預警計算場景

預警計算場景流程設計如下：

1. 未來第10s的數據(10萬條)與10s聚合的數據(10萬條）進行比對，值差異大於20% （即 ABS(預測值-真實值)/真實值大於0.2 ）列為異常數據。

1. 統計異常率（異常數據的比例）：異常數據數目/比對總數（10w） 。

1. 設定閾值0.215，當異常率超過閾值0.215時，進行報警。

5.實時異常率預警

5.1 處理流程

5.2 實現細節

5.2.1 實時數據模擬

用户可以利用下方代碼完成實時生產數據的模擬。可以通過調整變量 hour ,自定義生產多少小時數據。

由於數據為模擬數據，為了保證模型能夠完成數據擬合，這裏提前設定監測值與 5 個影響因素的關係為線性關係，以確保迴歸模型擬合的關係能夠收斂。

另外，為了模擬模型生成的預測值與真實值存在一定差異，方便進行異常比對，因此在代碼中，對線性參數進行了隨機化，以降低模型的準確率。用户真實場景中可自行調整參數方差大小，以控制模型準確率。簡言之，方差越小，模型擬合越高，模型精度越高。

orgColNames= `time`deviceCode`wind`humidity`air_pressure`temperature`life`propertyValue
orgColTypes = [TIMESTAMP,INT,DOUBLE,DOUBLE,DOUBLE,DOUBLE,INT,INT]
enableTableShareAndPersistence(table = streamTable(10000:0, orgColNames,orgColTypes), tableName=`dataTable, cacheSize = 6000000) 

def send_data(begintime,endtime,hour,devices_number,rate,mutable dest)
{

    btime=timestamp(begintime)
	do{
        seconds  = int(60*60*hour)  //定義需要壓入的批次，每秒鐘1批

        n = devices_number * rate* 1 // 每秒鐘生產10萬條數據
        time =sort(take(btime+(0..(1000-1)),n)) //
        deviceCode = take(1..devices_number,n)
        x1 = randNormal(25,2,n)  
        x2 = randNormal(55,5,n)
        x3 = randNormal(1.01325,0.00001,n)
        x4 = randNormal(75,5,n)
        x5 = int(randNormal(10,3,n))
        b1 = randNormal(0.4,0.05,n) //方差0.05 降低模型準確率
        b2 = randNormal(0.3,0.05,n)
        b3 = randNormal(0.2,0.05,n)
        b4 = randNormal(0.09,0.05,n)
        b5 = randNormal(0.01,0.001,n)
        bias = randNormal(5,1,n)
        propertyValue = int(b1*x1*10 + b2*x2*2 + b3*x3*1000 + b4*x4 + b5*x5 +bias)

        table_ps = table(time,deviceCode,x1,x2,x3,x4,x5,propertyValue)
        dest.append!(table_ps)

        btime=btime+1000
        etime=timestamp(now())
        timediff=btime-etime
        if(timediff>0){sleep(timediff)}
    }while(btime<=endtime)
}
hour = 0.5 //用户自定義壓入多少小時的數據,1 為 1個小時
devices_number = 100 //設備數目
rate = 1000 //每台設備 每秒鐘1000條數據
begintime = datetime(now()) //開始時間
endtime = begintime + int(hour * 3600-1)  //結束時間
submitJob(`send_data_to_kafka,`send_data_to_kafka,send_data,begintime,endtime,hour,devices_number,rate,dataTable) 

5.2.2 聚合計算

聚合計算通過 DolphinDB 內置的時間序列引擎 createTimeSeriesEngine 實現。訂閲數據表 dataTable 中的數據，將數據內容輸入聚合引擎 tradesAggregator 完成預警計算，最後將聚合計算結果存入聚合表 aggrTable 。

enableTableShareAndPersistence(table = streamTable(100000:0,orgColNames,orgColTypes), tableName=`aggrTable, cacheSize = 5000000) 

tradesAggregator = createTimeSeriesEngine(name="streamAggr", windowSize=10, step=10, metrics=<[avg(wind),avg(humidity),avg(air_pressure),avg(temperature),avg(life),avg(propertyValue)]>, dummyTable=dataTable, outputTable=aggrTable, timeColumn=`time, useSystemTime=false, keyColumn=`deviceCode, garbageSize=1000000)

subscribeTable(tableName="dataTable", actionName="Aggregator", offset=0, handler=append!{tradesAggregator}, msgAsTable=true)

5.2.3 數據持久化

訂閲數據表 dataTable 中的內容，將流表數據存入分佈式數據庫 dfs://Data 中。

/*
  * 將dataTable數據寫入分佈式表 

	數據建模：
	1）每小時記錄數：360,000,000
	2）每條記錄大小：46字節
	3）每小時空間佔用（壓縮前）：15.42G
	4）建議以“Id”值和“小時”進行組合分區，每分區≈157.93M
	5）分區索引為“時間戳”+“設備號”

  */
 def createConsumerDataBase(dbname,tbname,col_names,col_types){
    dbTime = database("",VALUE,datehour(2023.01.01T00:00:00)..datehour(2023.01.01T23:00:00))
    Ids = 1..100
    dbId = database("",VALUE,Ids)
    db = database(directory=dbname, partitionType=COMPO, partitionScheme=[dbTime,dbId],engine="TSDB")
    factor_partition = db.createPartitionedTable(table=table(1:0,col_names,col_types),tableName = tbname,partitionColumns = ["time","deviceCode"],sortColumns =["deviceCode","time"],compressMethods={time:"delta"},keepDuplicates=LAST)
}
dataTable_dbname,dataTable_tbname = "dfs://Data","data"
createConsumerDataBase(dataTable_dbname,dataTable_tbname,orgColNames,orgColTypes)
subscribeTable(tableName="dataTable", actionName="append_data_into_dfs", offset=0, handler=loadTable(dataTable_dbname,dataTable_tbname), msgAsTable=true,batchSize=100000, throttle=1, reconnect=true)

訂閲聚合表 aggrTable 中的內容，將流表數據存入分佈式數據庫 dfs://Aggregate 中。

/*
  * 將聚合計算結果寫入分佈式表
	數據建模：
		1）每小時記錄數：36,000,000
		2）每條記錄大小：46字節
		3）每小時空間佔用（壓縮前）：1.54G
		4）建議以“id”和“天”進行值分區，每分區≈ 379.03M
		5）分區索引為“時間戳”+“設備號”

  */
def createAggregateDataBase(dbname,tbname,col_names,col_types){
	if(existsDatabase(dbname)){dropDatabase(dbname)}
	Ids = 1..100
    dbId = database("",VALUE,Ids)
    dbTime = database("",VALUE,date(2023.01.01T00:00:00)..date(2023.12.31T20:00:00))
	db = database(directory=dbname, partitionType=COMPO, partitionScheme=[dbTime,dbId],engine="TSDB")
	factor_partition = db.createPartitionedTable(table=table(1:0,col_names,col_types),tableName = tbname,partitionColumns =["time","deviceCode"],sortColumns =["deviceCode","time"],compressMethods={time:"delta"},keepDuplicates=LAST)
}
aggr_dbname,aggr_tbname = "dfs://Aggregate","data"
createAggregateDataBase(aggr_dbname,aggr_tbname,orgColNames,orgColTypes)
subscribeTable(tableName="aggrTable", actionName="append_Aggregator_into_dfs", offset=0, handler=loadTable(aggr_dbname,aggr_tbname), msgAsTable=true,batchSize=100000, throttle=1, reconnect=true)

5.2.4 模型預測與預警計算

用户可以利用下方代碼完成模型預測與預警計算操作。

在進行模型預測與預警計算以前，需提前建立預測表 predictTable 與預警表 warningTable 以存儲預測數據與預警數據。

由於模型訓練需要用到10s數據，因此設定10s為一個週期，每10s 進行一次模型訓練、模型預測以及預警計算。

關於預警計算，設定預測值與聚合值的值差異大於20%時（即預測值大於真實值20%或預測值小於真實值20%）判定為異常值，當異常率大於0.215時進行預警。用户可根據需求自行調整。

//新建預測表
 preColNames = `time`deviceCode`wind`humidity`air_pressure`temperature`propertyValue_predicted
 preColTypes = orgColTypes
 enableTableShareAndPersistence(table = streamTable(100000:0,preColNames,preColTypes), tableName=`predictTable, cacheSize = 5000000)

 //新建預警表
 warnColNames = `time`abnormal_rate`whether_warning
 warnColTypes = [TIMESTAMP,DOUBLE,INT]
 enableTableShareAndPersistence(table = streamTable(100000:0,warnColNames,warnColTypes), tableName=`warningTable, cacheSize = 5000000)

/*
 * 根據 已有周期數據，對未來數據進行預測
 */
 def predictAndwarning(devices_number,rate,interval_past,interval_future,warning_threshold,mutable whether_start,DataTable_dfs,AggrTable_dfs,mutable predictTable,mutable warningTable){
    do{
      if(whether_start==false) {
        curtime = select top 1 time from DataTable_dfs  //第一次是從表中查詢最開始的時間
        curtime = curtime[`time][0]
        curtime_aggr = select top 1 time from AggrTable_dfs //每次儘量去與訓練數據的時間段同步
        curtime_aggr = curtime_aggr[`time][0]
        whether_start = true
      }
      curtime = curtime + interval_past*1000 //以後是直接往後推，最開始的時間往後推 10s中的間隔時間
      table_origin = select * from DataTable_dfs where time<=curtime and time>(curtime - interval_past*1000) //查詢當前時間前 interval 秒的數據
      if(table_origin.size()<interval_past*devices_number*rate) //如果查詢數據小於正常查詢到的數據數目
      {
        
         curtime = select top 1 time from DataTable_dfs order by time desc //從表中查詢最近的時間作為開始時間
         curtime = curtime[`time][0]
         table_origin = select * from DataTable_dfs where time<=curtime and time>(curtime - interval_past*1000) //查詢當前時間的前 interval 秒的數據
      }


      //訓練模型
      factors = sql(sqlCol([`wind,`humidity,`air_pressure,`temperature]), table_origin).eval()
      labels = table_origin[`propertyValue]
      model = knn(labels,factors,"regressor", 200);


      //模型預測
      curtime_aggr = curtime_aggr + interval_past*1000
      table_aggr = select * from AggrTable_dfs where time<=curtime_aggr and time>(curtime_aggr - interval_past*1000) order by deviceCode	 //查詢當前時間前 interval 秒的數據
      if(table_aggr.size()<interval_past*devices_number*rate/10) //如果查詢數據小於正常查詢到的數據數目
      {
         curtime_aggr = select top 1 time from AggrTable_dfs order by time desc //從表中查詢最近的時間作為開始時間
         curtime_aggr = curtime_aggr[`time][0]
         table_aggr = select * from AggrTable_dfs where time<=curtime_aggr and time>(curtime_aggr - interval_past*1000) //查詢當前時間的前 interval 秒的數據
      }
      pre_labels = sql(sqlCol([`wind,`humidity,`air_pressure,`temperature]), table_aggr).eval()
      //pre_values = predict(model,pre_labels)
      //暫時用隨機值代替///
      table_number =  table_aggr.size()
      x1 = randNormal(25,2,table_number)  //訓練數據
      x2 = randNormal(55,5,table_number)
      x3 = randNormal(1.01325,0.00001,table_number)
      x4 = randNormal(75,5,table_number)
      x5 = int(randNormal(10,3,table_number))
      b1 = randNormal(0.4,0.05,table_number) 
      b2 = randNormal(0.3,0.05,table_number)
      b3 = randNormal(0.2,0.05,table_number)
      b4 = randNormal(0.09,0.05,table_number)
      b5 = randNormal(0.01,0.001,table_number)
      bias = randNormal(5,1,table_number)
      propertyValue = int(b1*x1*10 + b2*x2*2 + b3*x3*1000 + b4*x4 + b5*x5 +bias)
      pre_values = propertyValue //暫時用隨機值代替
      /
    
      time =take(curtime_aggr + interval_future*1000+interval_future+(0..(1000-1)),table_number) //
      deviceCode = sort(take(1..devices_number,table_number))
      predicttempTable = table(time,deviceCode,pre_labels,pre_values as `propertyValue_predicted)
      predictTable.append!(predicttempTable) //預測結果導入流表

      //進行預警
      contrastTable = select propertyValue,propertyValue_predicted from lj(table_aggr,predicttempTable,`wind`humidity`air_pressure`temperature) //利用左連接
      abnormal_count = exec count(*) from contrastTable where propertyValue_predicted<0.8*propertyValue or propertyValue_predicted>1.2*propertyValue
      
      warning_time = curtime_aggr //進行預警的時間
      abnormal_rate = abnormal_count*1.0 / table_number
      whether_warning = 0 //默認不進行預警
      if(abnormal_rate>warning_threshold) whether_warning = 1 //當異常率超過閾值進行預警

      insert into warningTable values(warning_time,abnormal_rate, whether_warning);
      
      sleep(10000) //每10s進行一次預測
    }while(true)
}
devices_number = 100 //設備數目
rate = 1000 //每台設備 每秒鐘1000條數據
interval_past = 10 查詢過去10秒的數據
interval_future = 10 //預測未來第10秒的數據
warning_threshold = 0.215 //當異常值率大於0.215時進行預警
whether_start = false //標記
DataTable_dfs = loadTable(dataTable_dbname,dataTable_tbname)
AggrTable_dfs = loadTable(aggr_dbname,aggr_tbname)
submitJob(`predictAndwarning,`predictAndwarning,predictAndwarning,devices_number,rate,interval_past,interval_future,warning_threshold,whether_start,DataTable_dfs,AggrTable_dfs,predictTable,warningTable) //每10s 預測一次

5.3 Grafana 實時預警監控

Grafana 是由 Grafana Labs 公司開源的一個系統監測 ( System Monitoring ) 工具。

DolphinDB 開發了 Grafana 數據源插件，用户在 Grafana 面板上通過編寫查詢腳本、訂閲流數據表的方式，與 DolphinDB 進行交互，實現 DolphinDB 時序數據的可視化。具體操作可參考：DolphinDB Grafana DataSource Plugin 。

Grafana 中的 Query 代碼：

select * from warningTable

點擊 abnormal_rate，查看各時間段的異常率。

點擊 whether_warning，查看各時間段的預警情況。0不進行預警，1進行預警。

6. 總結

本教程以大型風電廠的發電量監控需求為例，利用 DolphinDB 內置機器學習框架中的 KNN 算法構建迴歸模型，實現了風機發電量的實時預警，從而幫助運維人員及時調整風機設備參數，更好地保障生產安全。此教程同樣可為物聯網行業中有數據模擬、機器學習模型訓練、實時預測、預警計算等需求的用户提供一份參考。

7. 附錄

7.1 開發環境

• 操作系統：Linux version 3.10.0-1160.el7.x86_64

• CPU: Intel(R) Xeon(R) Gold 5220R CPU @ 2.20GHz 64核

• 內存：503G

• 磁盤：SSD 3.5T

• 服務器端：DolphinDB 2.00.7

• 客户端：VSCode 1.73 （extension for DolphinDB v2.0.802）

7.2 開發腳本

環境清理腳本

//登錄
login(`admin,`123456)
version()

//取消後台任務
def cancelAllBatchJob(){
jobids=exec jobid from getRecentJobs() where endTime=NULL
cancelJob(jobids)
}
pnodeRun(cancelAllBatchJob)

t = getStreamingStat().pubTables
for(row in t){
	tableName = row.tableName
  if(row.actions[0]=='[') actions = split(substr(row.actions, 1, (strlen(row.actions)-2)), ",")
	else actions = split(substr(row.actions, 0, strlen(row.actions)), ",")
	for(action in actions){
		unsubscribeTable(tableName=tableName, actionName=action)
	}
}

//刪除流表
try{dropStreamTable(`dataTable)}catch(ex){}
try{dropStreamTable(`aggrTable)}catch(ex){}
try{dropStreamTable(`predictTable)}catch(ex){}
try{dropStreamTable(`warningTable)}catch(ex){}

//刪除計算引擎
try{dropStreamEngine("streamAggr")}catch(ex){}

//清理緩存
undef(all)
clearAllCache()

//清空數據庫，環境初始化
if(existsDatabase("dfs://Data")){dropDatabase("dfs://Data")}
if(existsDatabase("dfs://Aggregate")){dropDatabase("dfs://Aggregate")}
if(existsDatabase("dfs://predict")){dropDatabase("dfs://predict")}
if(existsDatabase("dfs://warning")){dropDatabase("dfs://warning")}

預測數據與預警數據持久化腳本

用途：將流表中的預測數據和預警數據持久化到分佈式表

//登錄
login(`admin,`123456)
version()

//取消後台任務
def cancelAllBatchJob(){
jobids=exec jobid from getRecentJobs() where endTime=NULL
cancelJob(jobids)
}
pnodeRun(cancelAllBatchJob)

t = getStreamingStat().pubTables
for(row in t){
	tableName = row.tableName
  if(row.actions[0]=='[') actions = split(substr(row.actions, 1, (strlen(row.actions)-2)), ",")
	else actions = split(substr(row.actions, 0, strlen(row.actions)), ",")
	for(action in actions){
		unsubscribeTable(tableName=tableName, actionName=action)
	}
}

//刪除流表
try{dropStreamTable(`dataTable)}catch(ex){}
try{dropStreamTable(`aggrTable)}catch(ex){}
try{dropStreamTable(`predictTable)}catch(ex){}
try{dropStreamTable(`warningTable)}catch(ex){}

//刪除計算引擎
try{dropStreamEngine("streamAggr")}catch(ex){}

//清理緩存
undef(all)
clearAllCache()

//清空數據庫，環境初始化
if(existsDatabase("dfs://Data")){dropDatabase("dfs://Data")}
if(existsDatabase("dfs://Aggregate")){dropDatabase("dfs://Aggregate")}
if(existsDatabase("dfs://predict")){dropDatabase("dfs://predict")}
if(existsDatabase("dfs://warning")){dropDatabase("dfs://warning")}

7.3 參考文獻

孫海麗, 龍翔, 韓蘭勝, 黃炎, 李清波. 《工業物聯網異常檢測技術綜述》. 通信學報, 2022(003):043.

裴丹, 張聖林, 裴昶華. 《基於機器學習的智能運維》. 中國計算機學會通訊, 2017, 013(012):68-72.

王加昌, 鄭代威, 唐雷, 鄭丹晨, 劉夢娟. 《基於機器學習的剩餘使用壽命預測實證研究》. 計算機科學, 2022, 49(11A): 211100285-9.