异常检测引擎

异常检测引擎由 createAnomalyDetectionEngine 函数创建。

DolphinDB 的异常检测引擎建立在流数据的发布-订阅模型之上。下例中，通过 createAnomalyDetectionEngine 创建异常检测引擎，并通过 subscribeTable 函数订阅流数据，每次有新数据流入就会按指定规则触发append!{engine}，将流数据持续输入聚合引擎（指定的数据表中）。异常检测引擎实时检测数据是否符合用户自定义的警报指标 temp>65，如发现异常数据，将它们输出到表 outputTable 中。

share streamTable(1000:0, `time`device`temp, [TIMESTAMP, SYMBOL, DOUBLE]) as sensor
share streamTable(1000:0, `time`device`anomalyType`anomalyString, [TIMESTAMP, SYMBOL, INT, SYMBOL]) as outputTable
engine = createAnomalyDetectionEngine("engine1", <[temp > 65]>, sensor, outputTable, `time, `device, 10, 1)
subscribeTable(, "sensor", "sensorAnomalyDetection", 0, append!{engine}, true)

这里对异常处理引擎涉及到的一些概念做简要介绍：

• 流数据表：DolphinDB 为流式数据提供的一种特定的表对象，提供流式数据的发布功能。通过subscribeTable函数，其他的节点或应用可以订阅和消费流数据。

• 异常处理引擎数据源：为异常处理引擎提供"原料"的通道。createAnomalyDetectionEngine函数返回一个抽象表，向这个抽象表写入数据，就意味着数据进入异常处理引擎进行计算。

• 异常指标：以元代码的格式提供一组处理流数据的布尔表达式。其中可以包含聚合函数，以支持复杂的场景。

• 数据窗口：每次计算时截取的流数据窗口长度。数据窗口仅在指标中包含聚合函数时有意义。

• 输出表：异常检测引擎的输出表第一列必须是时间类型，用于存放检测到异常的时间戳，如果有指定分组列，那么第二列为分组列，之后的两列分别为 int 类型和 string 或 symbol 类型，用于记录异常的类型（异常指标的表达式在 metrics 中的下标）和异常的内容。

异常检测引擎中的指标均要求返回布尔值。一般是一个函数或一个表达式。当指标中包含聚合函数，必须指定窗口长度和计算的时间间隔，每隔一段时间，在固定长度的移动窗口中计算指标。异常指标一般有以下三种类型：

• 只包含列名或非聚合函数，例如qty>10, lt(qty, prev(qty))。对于这类指标，异常检测引擎会对每一条收到的数据进行计算，判断是否符合指标并决定是否输出。

• 所有出现的列名都在聚合函数的参数中，例如avg(qty-price)>10, percentile(qty, 90)<100, sum(qty)>prev(sum(qty))。对于这类指标，异常检测引擎只会在窗口发生移动时对数据进行聚合计算，和时间序列聚合引擎（Time Series Aggregator）类似。

• 出现的列名中，既有作为聚合函数的参数的，又有不是聚合函数参数的，例如avg(qty) > qty, le(med(qty), price)。对于这类指标，异常检测引擎会在在窗口发生移动时进行聚合计算。每一条数据到达时，使用最近一个窗口的聚合函数的返回值与非聚合函数参数的列进行计算。

当异常指标中包含聚合函数时，用户必须指定数据窗口。流数据聚合计算是每隔一段时间，在固定长度的移动窗口中进行。窗口长度由参数 windowSize 设定；计算的时间间隔由参数 step 设定。

在有多组数据的情况下，若每组都根据各自第一条数据进入系统的时间来构造数据窗口的边界，则一般无法将各组的计算结果在相同数据窗口中进行对比。考虑到这一点，系统按照参数 step 值确定一个整型的规整尺度 alignmentSize ，以对各组第一个数据窗口的边界值进行规整处理。

（1）当数据时间类型为 MONTH 时，会以第一条数据对应年份的 1 月作为窗口的上边界。

（2）当数据的时间类型为 DATE 时，不对第一个数据窗口的边界值进行规整。

（3）当数据时间精度为秒或分钟时，如 MINUTE, DATETIME 或 SECOND类型，alignmentSize 取值规则如下表：

如果 roundTime = false，对于 step > 60，alignmentSize 都为60。 如果 roundTime = true，则 alignmentSize 取值规则如下表：

（4）当数据时间精度为毫秒时，如 TIMESTAMP 或 TIME 类型，alignmentSize 取值规则如下表：

如果 roundTime = false，对于 step > 60000，alignmentSize 都为60000。 如果 roundTime = true，则 alignmentSize 取值规则如下表：

假设第一条数据时间的最小精度值为 x，那么第一个数据窗口的左边界最小精度经过规整后为x/alignmentSize\*alignmentSize，其中/代表相除后取整。举例来说，若第一条数据时间为 2018.10.08T01:01:01.365，则 x=365。若 step=100，根据上表，alignmentSize=100，可得出规整后的第一个数据窗口左边界最小精度为 365\100*100=300，因此规整后的第一个数据窗口范围为 2018.10.08T01:01:01.300 至 2018.10.08T01:01:01.400。

快照机制可以用于系统出现异常之后，对引擎进行恢复。通过以下这个例子，可以理解 snapshotDir 和 snapshotIntervalInMsgCount 的作用。如果启用snapshot，引擎订阅流表时，handler 需是 appendMsg 函数，需指定 handlerNeedMsgId = true，用来记录快照的消息位置。

WORK_DIR="/home/root/WORK_DIR"
mkdir(WORK_DIR+"/snapshotDir")
enableTableShareAndPersistence(table = streamTable(10000:0,`time`sym`price`qty, [TIMESTAMP,SYMBOL,DOUBLE,INT]) , tableName="trades", cacheSize=1000000)
enableTableShareAndPersistence(table = streamTable(10000:0, `time`sym`type`metric, [TIMESTAMP,STRING,INT,STRING]), tableName = "output", cacheSize=1000000)
go

adengine = createAnomalyDetectionEngine(name="test", metrics=<[avg(qty)>1]>, dummyTable=trades, outputTable=output, timeColumn=`time, keyColumn=`sym, windowSize=10, step=10, snapshotDir=WORK_DIR+"/snapshotDir", snapshotIntervalInMsgCount=100)
subscribeTable(server="", tableName="trades", actionName="adengine",offset= 0, handler=appendMsg{adengine}, msgAsTable=true, handlerNeedMsgId=true)

def writeData(mutable t){
do{
batch = 10
tmp = table(batch:batch, `time`sym`price`qty, [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE])
tmp[`time] = take(now(), batch)
tmp[`sym] = "A"+string(1..batch)
tmp[`price] = round(rand(100.0, batch), 2)
tmp[`qty] = rand(10, batch)
t.append!(tmp)
sleep(1000)
}while(true)
}

job1=submitJob("write", "", writeData, trades)
//执行一段时间后重启server

enableTableShareAndPersistence(table = streamTable(10000:0,`time`sym`price`qty, [TIMESTAMP,SYMBOL,DOUBLE,INT]) , tableName="trades", cacheSize=1000000)
enableTableShareAndPersistence(table = streamTable(10000:0, `time`sym`type`metric, [TIMESTAMP,STRING,INT,STRING]), tableName = "output", cacheSize=1000000)

select last(time) from output
>2021.03.16T11:59:10.920

select last(time) from trades
>2021.03.16T11:59:13.916

WORK_DIR="/home/root/WORK_DIR"
adengine = createAnomalyDetectionEngine(name="test", metrics=<[avg(qty)>qty]>, dummyTable=trades, outputTable=output, timeColumn=`time, keyColumn=`sym, windowSize=10, step=10, snapshotDir=WORK_DIR+"/snapshotDir", snapshotIntervalInMsgCount=100)

ofst = getSnapshotMsgId(adengine)
print(ofst)
>299

select count(*) from trades
>390

//从第300条数据开始订阅
subscribeTable(server="", tableName="trades", actionName="adengine",offset=ofst+1, handler=appendMsg{adengine}, msgAsTable=true, handlerNeedMsgId=true) 

DolphinDB 提供的异常检测引擎是一个轻量、使用方便的流数据引擎，它通过与流数据表合作来完成流数据的实时检测任务，能够满足物联网实时监控和预警的需求。