云边协同：DolphinDB 构建一体化振动监控与故障诊断解决方案

创建如下4个分布式表，其中 warn 表用于存储异常告警信息，wave 表用于存储波形的峰值趋势信息，alterWave 表用于存储异常发生时的异常波形，wave5M 则用于存储降采样后的数据。

//创建 warn 分布式表
db = database("dfs://warnDB", VALUE, 2022.01.01..2022.12.31)
m = table(1:0,`datetime`source`type`metric,[TIMESTAMP,SYMBOL,INT,STRING]) 
db.createPartitionedTable(m, "warn", ["datetime"])

//创建聚合结果分布式表
db = database("dfs://waveDB", VALUE, 2022.01.01..2022.12.31)
m = table(1:0, `datetime`source`max_wave`min_wave,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE]) 
db.createPartitionedTable(m, "wave", ["datetime"])

//创建异常录波表
db = database("dfs://alterWaveDB", VALUE, 2022.01.01..2022.12.31)
m = table(1:0,`datetime`source`wave_datatime <- (`d+string(0..255)),[TIMESTAMP,SYMBOL,TIMESTAMP] <- take(DOUBLE,256))
db.createPartitionedTable(m, "alterWave", ["datetime"])

//创建降采样后的原始数据存储表
db = database("dfs://waveDB5M", VALUE, 2022.01.01..2022.12.31)
m = table(1:0,`datetime`source <- (`d+string(0..255)),[TIMESTAMP,SYMBOL] <- take(DOUBLE,256))
db.createPartitionedTable(m, "wave5M", ["datetime"])

share streamTable(100:0, `datetime`source <- (`d+string(0..255)),[TIMESTAMP,SYMBOL] <- take(DOUBLE,256)) as wave
share streamTable(100:0, `datetime`source <- (`d+string(0..255)),[TIMESTAMP,SYMBOL] <- take(DOUBLE,256)) as wave5M
share streamTable(100:0, `datetime`source`max_wave`min_wave,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE]) as stwave
share streamTable(100:0, `datetime`source`type`metric,[TIMESTAMP,SYMBOL,INT,STRING]) as warn
share streamTable(100:0, `datetime`source`wave_datatime <- (`d+string(0..255)),[TIMESTAMP,SYMBOL,TIMESTAMP] <- take(DOUBLE,256)) as windowOutput1

上述代码创建了五个流表，其中 wave 流表用于接收解析后的振动数据，wave5M 用于转存降采样后的数据，stwave 存储波形峰值，warn 用于存储异常记录，windowOutput1 用于存储异常波形数据。

数据处理部分包括了数据降采样、峰值趋势计算、异常检测和异常检测录波四部分内容。

• 使用时间序列引擎，对原数据进行降采样

首先是对数据进行降采样，通过时间序列引擎，进行五分钟频率的降采样，降低边缘端的数据存储压力。

//创建时序引擎，每五分钟保留最后一条原始数据，对原数据进行降采样
metricsDownSampling = parseExpr(each(add,'last(',(`d+string(0..255)+")")))
downsampling = createTimeSeriesEngine(name="downsampling",  windowSize=5*60*1000, step=5*60*1000, metrics=metricsDownSampling, dummyTable=wave, outputTable=wave5M, timeColumn=`datetime, keyColumn=`source)
subscribeTable(tableName='wave',actionName="down_sampling", offset=0, handler=append!{downsampling}, msgAsTable=true,batchSize=10000,throttle=1)
subscribeTable(tableName='wave5M',actionName='save5mToDFS',offset=0,handler=append!{loadTable("dfs://waveDB5M",'wave5M')},msgAsTable=true,batchSize=10000,throttle=1)

上述代码创建了一个时间序列引擎，时间窗口长度及移动步长均为5分钟。该时间序列引擎会将输入数据按时间进行进行聚合，每5分钟聚合一次并返回最后一条记录存储到 wave5M 表中。同时创建了两个订阅：

1. down_sampling 订阅 wave 流表，并将其注入时间序列引擎中进行降采样
2. save5mToDFS 订阅 wave5M 流表，将降采样后数据存入分布式库表中

• 使用时间序列引擎，计算每秒波形峰值

此处我们仍然使用时间序列引擎，计算每秒的波形峰值，通过保留波形的峰值来记录数据的整体特征及趋势。

//定义处理函数
def metricCal(engineName,msg){
	t = select datetime,source,rowMin(msg[,2:258]) as min,rowMax(msg[,2:258]) as max  from msg
    getStreamEngine(engineName).append!(t)
}
//定义聚合规则
metrics=<[max(max),min(min)]>
inputSchema = table(100:0, `datetime`source`min`max,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE])
//创建时序引擎，窗口长度为1s，步长为1s
tsAggr1 = createTimeSeriesEngine(name="tsAggr1",  windowSize=1000, step=1000, metrics=metrics, dummyTable=inputSchema, outputTable=stwave, timeColumn=`datetime, keyColumn=`source)
//订阅 wave 表，进行指标计算
subscribeTable(tableName="wave", actionName="act_tsAggr1", offset=0, handler=metricCal{"tsAggr1"}, msgAsTable=true,batchSize=10000,throttle=1);
//订阅 stwave 表，将峰值趋势入库
subscribeTable(tableName='stwave',actionName='savePeakToDFS',offset=0,handler=append!{loadTable("dfs://waveDB",'wave')},msgAsTable=true,batchSize=10000,throttle=1)

上述代码有如下几个数据处理流程：

1. 首先订阅 wave 流表，计算每一行测点的最大值和最小值，将计算结果输入时间序列引擎 tsAggr1 中
2. tsAgg1 引擎按秒聚合输出每秒波形数据的峰值，存储到 stwave 流表中。
3. 最后订阅 stwave 表，将峰值趋势数据存入分布式数据库表。

• 创建异常检测引擎，进行异常检测

//创建异常检测引擎，检测规则 max>90, min<-90
tsAggr2 = createAnomalyDetectionEngine(name="tsAggr2", metrics=<[max_wave > 90, min_wave <-90]>, dummyTable=stwave, outputTable=warn, timeColumn=`datetime, keyColumn=`source, windowSize=60*1000, step=60*1000)
//订阅 stwave 进行异常检测
subscribeTable(tableName="stwave", actionName="act_tsAggr2", offset=0, handler=append!{tsAggr2}, msgAsTable=true,batchSize=10000,throttle=1);
//订阅 warn 表，将异常记录入库
subscribeTable(tableName='warn',actionName='saveWarnToDFS',offset=0,handler=append!{loadTable("dfs://warnDB",'warn')},msgAsTable=true,batchSize=10000,throttle=1)

创建异常检测引擎，并订阅峰值流表 stwave，将异常结果(max_wave > 90, min_wave <-90)输入异常记录表 warn 中，同时订阅 warn 表将异常数据存入分布式数据库表中。

• 创建 windowjoin 引擎，订阅 warn 和 wave，录制异常波形

//创建 windowjoin 引擎，订阅 warn 和 wave，输出待存数据
cbefore = short(1)*(-1000)
cafter = short(1)*1000
metricswj = parseExpr(("wave.datetime" <- (`d+string(0..255))))
wjengine = createWindowJoinEngine(name="waveStreamWindow",leftTable=warn,rightTable=wave,outputTable=windowOutput1,window=cbefore:cafter,
                                  metrics=metricswj,matchingColumn=`source,timeColumn=`datetime)
subscribeTable(tableName=`warn,actionName="joinLeft",offset=0,handler=appendForJoin{wjengine,true},msgAsTable=true,batchSize=10000,throttle=1)
subscribeTable(tableName=`wave,actionName="joinRight",offset=0,handler=appendForJoin{wjengine,false},msgAsTable=true,batchSize=10000,throttle=1)  
//订阅 windowOutput1，异常波形入库
subscribeTable(tableName='windowOutput1',actionName='saveAlterWaveToDFS',offset=0,handler=append!{loadTable("dfs://alterWaveDB",'alterWave')},msgAsTable=true,batchSize=10000,throttle=1)                                 

订阅 warn 和 wave，使用窗口连接引擎，以异常记录时间 t 为中心，选择 [t-1,t+1] 的波形数据作为异常波形数据，输入 windowOutput1 中，订阅 windowOutput1，并存入分布式数据库表中。

本文使用脚本模拟4个通道，256个测点数据的实时写入，采样频率为 50HZ，具体脚本见附件 simuateData 函数。

注： 本文模拟生成的数据完全随机，故多次计算结果可能不同。

查看降采样后数据，可以看到每个通道都进行了5分钟降采样，共计12条记录：

wave5M= loadTable('dfs://waveDB5M','wave5M')
select * from wave5M

查看峰值趋势表，每秒保留该通道的最大及最小值：

wave= loadTable('dfs://waveDB','wave')
select * from wave

查看异常告警表输出：

warn= loadTable('dfs://warnDB','warn')
select * from warn

波形录制表为根据异常告警表与原始数据表连接的结果，因此异常波形表的结果应与告警表一致：

alterWave= loadTable('dfs://alterWaveDB','alterWave')
select max(wave_datatime),min(wave_datatime) from alterWave group by datetime,source

可以看到异常发生后的前后一秒的数据都被记录了下来。