DolphinDB异常检测引擎：实时告警

摘要

本文深入讲解DolphinDB异常检测引擎。从引擎原理到创建配置，从规则定义到告警输出，从多规则组合到实战应用，全面介绍异常检测引擎的核心功能。通过丰富的代码示例，帮助读者掌握实时告警系统的核心技能。

一、异常检测引擎概述

1.1 什么是异常检测引擎

异常检测引擎是DolphinDB内置的流计算引擎，用于实时检测数据异常：

1.2 异常检测引擎特点

1.3 适用场景

二、创建异常检测引擎

2.1 基本语法

//创建异常检测引擎 agg=createAnomalyDetectionEngine("engine_name",//引擎名称 metrics,//异常检测规则（Lambda表达式列表） outputTable,//输出表 timeColumn,//时间列[keyColumn],//分组列（可选）[windowSize],//窗口大小[garbageSize]//垃圾回收阈值)

2.2 创建简单引擎

//创建输入流表 share streamTable(1:0,`device_id`timestamp`temperature,[INT,TIMESTAMP,DOUBLE])asinput_stream//创建输出表 share table(1:0,`device_id`timestamp`temperature`anomaly_type,[INT,TIMESTAMP,DOUBLE,SYMBOL])asoutput_table//创建异常检测引擎 agg=createAnomalyDetectionEngine("anomaly_engine",<[temperature>30,//规则1：温度>30temperature<10]>,//规则2：温度<10output_table,`timestamp,`device_id)//订阅流表 subscribeTable(,"input_stream","anomaly_detect",-1,agg,true)

2.3 输出格式

//异常检测输出格式//-device_id:分组列//-timestamp:时间列//-temperature:检测列//-anomaly_type:异常类型（规则索引）//例如：//device_id|timestamp|temperature|anomaly_type//1|10:00:00|35.0|0(规则0匹配：温度>30)//2|10:00:01|5.0|1(规则1匹配：温度<10)

三、异常检测规则

3.1 阈值规则

//阈值规则：简单的阈值判断 agg=createAnomalyDetectionEngine("threshold_engine",<[temperature>30,//高温告警 temperature<10,//低温告警 humidity>80,//高湿度告警 humidity<20]>,//低湿度告警 output_table,`timestamp,`device_id)

3.2 统计规则

//统计规则：基于统计指标//需要配合窗口使用 agg=createAnomalyDetectionEngine("stats_engine",<[temperature>avg(temperature)+3*std(temperature),//超过3倍标准差 temperature<avg(temperature)-3*std(temperature)]>,//低于3倍标准差 output_table,`timestamp,`device_id,60000)//窗口大小60秒

3.3 组合规则

//组合规则：多条件组合 agg=createAnomalyDetectionEngine("combo_engine",<[temperature>30andhumidity>70,//高温高湿 temperature<10andhumidity<30,//低温低湿 temperature>35orhumidity>90]>,//高温或高湿 output_table,`timestamp,`device_id)

3.4 复杂规则

//复杂规则：使用函数 agg=createAnomalyDetectionEngine("complex_engine",<[abs(temperature-prev(temperature))>5,//温度突变>5度 temperature>30andtemperature>prev(temperature)*1.2]>,//温度突增20%output_table,`timestamp,`device_id)

四、窗口模式

4.1 无窗口模式

//无窗口：每条数据独立判断 agg=createAnomalyDetectionEngine("no_window_engine",<[temperature>30]>,output_table,`timestamp,`device_id)//适用场景：//-简单阈值判断//-不需要历史数据

4.2 有窗口模式

//有窗口：基于窗口内数据判断 agg=createAnomalyDetectionEngine("window_engine",<[temperature>avg(temperature)+3*std(temperature)]>,output_table,`timestamp,`device_id,60000)//60秒窗口//适用场景：//-统计规则//-需要历史数据

五、实战案例

5.1 设备温度告警系统

//==========1.创建流表==========share streamTable(100000:0,`device_id`timestamp`temperature`humidity`pressure,[INT,TIMESTAMP,DOUBLE,DOUBLE,DOUBLE])assensor_stream//==========2.创建告警输出表==========share table(1:0,`device_id`timestamp`value`anomaly_type`alert_level,[INT,TIMESTAMP,DOUBLE,SYMBOL,INT])asalert_table//==========3.创建异常检测引擎==========agg=createAnomalyDetectionEngine("temp_alert_engine",<[temperature>35,//规则0：严重高温 temperature>30,//规则1：一般高温 temperature<5,//规则2：严重低温 temperature<10]>,//规则3：一般低温 alert_table,`timestamp,`device_id)//==========4.订阅流表==========subscribeTable(,"sensor_stream","temp_alert",-1,agg,true)//==========5.添加告警级别==========share table(1:0,`device_id`timestamp`value`anomaly_type`alert_level,[INT,TIMESTAMP,DOUBLE,SYMBOL,INT])asfinal_alert_table subscribeTable(,"alert_table","level_handler",-1,def(msg){result=select device_id,timestamp,value,anomaly_type,casewhen anomaly_typein["0","2"]then2//严重else1endasalert_levelfrommsg final_alert_table.append!(result)},true)//==========6.模拟数据==========defsimulateAlert(){for(iin1..100){sensor_stream.append!(table(take(1..10,100)asdevice_id,take(now(),100)astimestamp,rand(0.0..40.0,100)astemperature,//包含异常温度 rand(40.0..60.0,100)ashumidity,rand(1000.0..1020.0,100)aspressure))sleep(100)}}simulateAlert()//查看告警 select*fromfinal_alert_table order by timestamp desc limit20

5.2 多指标组合告警

//==========1.创建流表==========share streamTable(100000:0,`device_id`timestamp`temperature`humidity`vibration,[INT,TIMESTAMP,DOUBLE,DOUBLE,DOUBLE])assensor_stream//==========2.创建告警输出表==========share table(1:0,`device_id`timestamp`anomaly_type`description,[INT,TIMESTAMP,SYMBOL,STRING])ascombo_alert_table//==========3.创建组合告警引擎==========agg=createAnomalyDetectionEngine("combo_alert_engine",<[temperature>30andhumidity>70,//高温高湿 temperature>30andvibration>5,//高温高振动 humidity>80andvibration>3,//高湿高振动 temperature>35orvibration>10]>,//严重异常 combo_alert_table,`timestamp,`device_id)//==========4.订阅流表==========subscribeTable(,"sensor_stream","combo_alert",-1,agg,true)

5.3 统计异常检测

//==========1.创建流表==========share streamTable(100000:0,`device_id`timestamp`temperature,[INT,TIMESTAMP,DOUBLE])assensor_stream//==========2.创建告警输出表==========share table(1:0,`device_id`timestamp`temperature`avg_temp`std_temp`anomaly_type,[INT,TIMESTAMP,DOUBLE,DOUBLE,DOUBLE,SYMBOL])asstats_alert_table//==========3.创建统计异常引擎==========agg=createAnomalyDetectionEngine("stats_alert_engine",<[temperature>avg(temperature)+3*std(temperature),temperature<avg(temperature)-3*std(temperature)]>,stats_alert_table,`timestamp,`device_id,300000)//5分钟窗口//==========4.订阅流表==========subscribeTable(,"sensor_stream","stats_alert",-1,agg,true)

六、引擎管理

6.1 查看引擎状态

//查看所有引擎状态 getStreamEngineStat()//查看特定引擎 getStreamEngineStat("anomaly_engine")

6.2 删除引擎

//删除引擎 dropStreamEngine("anomaly_engine")

6.3 引擎监控

//引擎监控函数defmonitorAnomalyEngine(){stat=getStreamEngineStat()for(rowinstat){if(row.type=="AnomalyDetectionEngine"){print("异常检测引擎: "+row.name)print(" 状态: "+row.status)print(" 规则数: "+string(row.numMetrics))print(" 处理行数: "+string(row.processedRows))print(" 告警数: "+string(row.outputRows))}}}monitorAnomalyEngine()

七、告警通知

7.1 告警推送

//告警推送函数defpushAlert(alert){//发送邮件//sendEmail(alert)//发送短信//sendSMS(alert)//发送微信//sendWeChat(alert)print("告警推送: "+alert.device_id+" - "+alert.anomaly_type)}//订阅告警表 subscribeTable(,"alert_table","push_handler",-1,def(msg){for(rowinmsg){pushAlert(row)}},true)

7.2 告警聚合

//告警聚合：避免告警风暴 share table(1:0,`device_id`anomaly_type`first_time`last_time`count,[INT,SYMBOL,TIMESTAMP,TIMESTAMP,LONG])asaggregated_alerts//聚合逻辑defaggregateAlerts(msg){for(rowinmsg){existing=select*fromaggregated_alerts where device_id=row.device_idandanomaly_type=row.anomaly_typeandlast_time>now()-60000//1分钟内if(existing.rows()>0){update aggregated_alertssetlast_time=row.timestamp,count=count+1where device_id=row.device_idandanomaly_type=row.anomaly_typeandlast_time>now()-60000}else{insert into aggregated_alerts values(row.device_id,row.anomaly_type,row.timestamp,row.timestamp,1)}}}subscribeTable(,"alert_table","aggregate_handler",-1,aggregateAlerts,true)

八、总结

本文详细介绍了DolphinDB异常检测引擎：

1. 引擎原理：实时异常检测
2. 创建方法：简单引擎、分组引擎
3. 检测规则：阈值规则、统计规则、组合规则
4. 窗口模式：无窗口、有窗口
5. 实战应用：温度告警、组合告警、统计异常
6. 告警通知：告警推送、告警聚合

思考题：

1. 如何设计合理的告警规则？
2. 如何避免告警风暴？
3. 统计异常检测适合什么场景？

参考资料

• DolphinDB异常检测引擎
• DolphinDB流计算