一、场景与需求

1.1 业务场景

先看一个典型的工业场景：一条产线上有50台设备，每台设备装有温度和振动两个传感器。数据每秒采集一次，也就是每秒产生100条记录（50台×2个测点）。业务上要求的是：

• 当某台设备的振动值超过阈值时，实时触发告警
• 当某台设备的温度在30秒内持续上升超过5°C时，触发告警
• 当某台设备的振动值在过去60秒内的标准差突然增大（偏离历史基线）时，触发告警

1.2 异常检测引擎

传统做法往往需要这么一套链路：数据先写入时序数据库，然后经过Kafka消费，再到Flink做流处理，最后把告警写入或推送给下游。组件多、链路长，维护成本不低。

DolphinDB的做法是把流计算引擎直接内置在数据库里。数据写入的同时就能触发检测逻辑，端到端延迟控制在毫秒级。更关键的是，流计算和批处理共用同一套脚本语言——研发阶段可以用历史数据验证规则，验证通过后直接切到流数据上线，一句代码都不用改。这种“流批一体”的设计，开发到上线的周期可以从“周”缩短到“小时”。

二、数据准备：模拟工业传感器数据

2.1 创建存储表

先定义一张分布式表来存储传感器数据：

// 创建数据库（按日期   设备ID组合分区）
dbName = "dfs://iot_monitor"
if (existsDatabase(dbName)) dropDatabase(dbName)
db1 = database(, VALUE, 2024.01.01..2024.12.31)
db2 = database(, HASH, [SYMBOL, 10])
db = database(dbName, COMPO, [db1, db2])

// 定义表结构并创建分布式表
schema = table(1:0, `deviceId`timestamp`metricType`value, [SYMBOL, TIMESTAMP, SYMBOL, DOUBLE])
db.createPartitionedTable(schema, `sensorData, `timestamp`deviceId)

2.2 模拟数据生成函数

写一个函数来模拟传感器数据。正常运行下，温度在40-60°C之间波动，振动值在0.5-2.0g之间波动。我们可以注入异常来测试检测引擎的效果：

def simulateSensorData(deviceCount, duration, anomalyDevice, anomalyTime) {
    devices = "device_"   string(1..deviceCount)
    t = table(deviceCount * 2 * duration * 1000:0,
        `deviceId`timestamp`metricType`value,
        [SYMBOL, TIMESTAMP, SYMBOL, DOUBLE])
    startTs = now() - duration * 1000
    for (ts in startTs..(startTs   (duration - 1) * 1000)) {
        for (dev in devices) {
            // 温度：正常 40-60°C，正弦波 随机噪声
            tempBase = 50   10 * sin(double(ts % 86400000) / 86400000 * 2 * pi)
            tempNoise = norm(0, 2)
            tempValue = tempBase   tempNoise
            // 振动：正常 0.5-2.0g，随机波动
            vibBase = 1.0   norm(0, 0.3)
            // 注入异常：指定设备在指定时间后出现异常
            isAnomaly = (dev == anomalyDevice) && (ts >= anomalyTime)
            if (isAnomaly) {
                vibValue = vibBase   4.0   norm(0, 0.5)  // 振动值飙升
                tempValue = tempValue   15                // 温度飙升
            } else {
                vibValue = vibBase
            }
            insert into t values(dev, timestamp(ts), `temperature, tempValue)
            insert into t values(dev, timestamp(ts), `vibration, vibValue)
        }
    }
    return t
}

2.3 生成测试数据并写入

生成10分钟的模拟数据，在第5分钟对device_3注入异常：

// 生成数据：50台设备，600秒（10分钟），device_3在第300秒开始异常
testData = simulateSensorData(50, 600, "device_3", now() - 300 * 1000)

// 写入分布式表
loadTable("dfs://iot_monitor", "sensorData").append!(testData)

// 查看数据概况
select count(*) as totalRows, min(timestamp) as startTs, max(timestamp) as endTs
from loadTable("dfs://iot_monitor", "sensorData")

三、用历史数据验证检测规则

规则正式部署到流计算引擎之前，先用历史数据批量验证一下。这是流批一体最实在的优势——同一套逻辑，先批处理跑一遍，没问题再流式上线。

3.1 规则一：振动值超过阈值

最简单的阈值检测：

// 查询振动值超过 4.0g 的记录
select * from loadTable("dfs://iot_monitor", "sensorData")
where metricType = `vibration
and value > 4.0
and timestamp >= now() - 600 * 1000
order by timestamp

预期结果：只有device_3在第5分钟之后会出现超过阈值的记录。

3.2 规则二：温度30秒内持续上升超过5°C

利用DolphinDB的滑动窗口函数deltas计算温度变化量：

// 按设备分组，计算30秒窗口内的温度变化量
select deviceId, timestamp, value as temp, deltas(value, 30) as tempChange30s
from loadTable("dfs://iot_monitor", "sensorData")
where metricType = `temperature
and timestamp >= now() - 600 * 1000
context by deviceId
ha ving deltas(value, 30) > 5

3.3 规则三：振动标准差突然增大

计算滑动窗口标准差，并与历史基线比较：

// 计算60秒滑动窗口标准差，基线为前300秒的均值
select deviceId, timestamp, value as vibration,
    mstd(value, 60) as vibStd60s,
    ma vg(mstd(value, 60), 300) as baselineStd,
    mstd(value, 60) - ma vg(mstd(value, 60), 300) as stdDeviation
from loadTable("dfs://iot_monitor", "sensorData")
where metricType = `vibration
and timestamp >= now() - 600 * 1000
context by deviceId
ha ving abs(mstd(value, 60) - ma vg(mstd(value, 60), 300)) > 0.5

三条规则都能准确检出device_3的异常行为，验证通过。

四、创建流计算异常检测引擎

规则验证没问题，接下来把同样的逻辑部署到流计算引擎，实现实时检测。

4.1 创建流数据表

// 清理已有流表（如果存在）
try { undef(`streamSensor, SHARED) } catch(ex) {}
// 创建共享流表，用于接收实时写入的传感器数据
share streamTable(100000:0, `deviceId`timestamp`metricType`value,
    [SYMBOL, TIMESTAMP, SYMBOL, DOUBLE]) as streamSensor

4.2 创建告警输出表

try { undef(`alerts, SHARED) } catch(ex) {}
share streamTable(10000:0, `alertTime`deviceId`alertType`alertDetail`severity,
    [TIMESTAMP, SYMBOL, SYMBOL, STRING, SYMBOL]) as alerts

4.3 引擎一：简单阈值检测

// 清理已有引擎
try { dropStreamEngine("thresholdDetector") } catch(ex) {}
// 创建异常检测引擎：振动值 > 4.0 触发告警
thresholdDetector = createAnomalyDetectionEngine(name = "thresholdDetector",
    metrics = [ 4.0>],
    dummyTable = streamSensor,
    outputTable = alerts,
    timeColumn = `timestamp,
    keyColumn = `deviceId,
    windowSize = 1000, // 1秒窗口
    step = 1000        // 1秒步长
)
// 订阅流表，将数据注入检测引擎
subscribeTable(tableName="streamSensor", actionName="thresholdDetect", offset=0, handler=append!{thresholdDetector}, msgAsTable=true)

4.4 引擎二：时间序列聚合检测

对于“温度30秒内上升超过5°C”和“振动标准差偏离基线”这类需要窗口计算的规则，使用时间序列聚合引擎：

try { dropStreamEngine("tsAggDetector") } catch(ex) {}
// 先创建一个中间结果表，存储窗口聚合指标
try { undef(`aggMetrics, SHARED) } catch(ex) {}
share streamTable(100000:0, `deviceId`aggTime`vibMax`tempDelta`vibStd`,
    [SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE]) as aggMetrics

tsAggDetector = createTimeSeriesEngine(name = "tsAggDetector",
    windowSize = 30000,  // 30秒窗口
    step = 5000,         // 5秒输出一次
    metrics = [, , ],
    dummyTable = streamSensor,
    outputTable = aggMetrics,
    timeColumn = `timestamp,
    keyColumn = `deviceId,
    useSystemTime = false)
subscribeTable(tableName="streamSensor", actionName="tsAggDetect", offset=0, handler=append!{tsAggDetector}, msgAsTable=true)

然后对聚合结果做二次判断，将告警写入输出表：

// 对聚合指标表做条件过滤，写入告警表
def checkAggAlerts(metrics) {
    alerts = select now() as alertTime, deviceId,
        iif(vibMax > 4.0, "vibration_threshold", "")   
        iif(tempDelta > 5, "temperature_surge", "") as alertType,
        "vibMax="   string(vibMax)   ", tempDelta="   string(tempDelta) as alertDetail,
        iif(vibMax > 6.0 or tempDelta > 8, "critical", "warning") as severity
    from metrics
    where vibMax > 4.0 or tempDelta > 5 or vibStd > 1.5
    // 只有真正有告警时才输出
    if (alerts.size() > 0) {
        getStreamAlert().append!(alerts)
    }
}
def getStreamAlert() { return alerts }
// 订阅聚合指标表
subscribeTable(tableName="aggMetrics", actionName="checkAlerts", offset=0, handler=checkAggAlerts, msgAsTable=true)

五、实时测试与效果验证

5.1 模拟实时数据写入

// 写入正常数据（5秒的量）
def writeNormalBatch() {
    devices = "device_"   string(1..50)
    t = table(500:0, `deviceId`timestamp`metricType`value,
        [SYMBOL, TIMESTAMP, SYMBOL, DOUBLE])
    ts = now()
    for (i in 1..5) {
        for (dev in devices) {
            insert into t values(dev, timestamp(ts   i*1000), `temperature, 50   norm(0, 2))
            insert into t values(dev, timestamp(ts   i*1000), `vibration, 1.0   norm(0, 0.3))
        }
    }
    streamSensor.append!(t)
}
writeNormalBatch()
// 检查告警表——应该为空
select * from alerts

// 写入异常数据：device_3 振动值飙升
def writeAnomalyBatch() {
    t = table(10:0, `deviceId`timestamp`metricType`value,
        [SYMBOL, TIMESTAMP, SYMBOL, DOUBLE])
    ts = now()
    for (i in 1..10) {
        insert into t values("device_3", timestamp(ts   i*1000), `vibration, 5.5   norm(0, 0.5))
        insert into t values("device_3", timestamp(ts   i*1000), `temperature, 70   norm(0, 3))
    }
    streamSensor.append!(t)
}
writeAnomalyBatch()
// 等待1秒后查看告警
timer(1000) { /* 等待引擎处理 */ }
select * from alerts order by alertTime desc

预期输出：

alertTime           deviceId  alertType               alertDetail                              severity
2024-06-06 10:30:15 device_3  vibration_threshold     vibMax=6.12, tempDelta=0                critical
2024-06-06 10:30:15 device_3  vibration_threshold     vibMax=5.83, tempDelta=0                warning
...

5.2 验证检测延迟

// 测量端到端延迟：从数据写入到告警输出的时间差
def measureLatency() {
    // 写入带时间戳的测试数据
    writeTs = now()
    t = table(1:0, `deviceId`timestamp`metricType`value,
        [SYMBOL, TIMESTAMP, SYMBOL, DOUBLE])
    insert into t values("device_test", timestamp(writeTs), `vibration, 9.9)
    streamSensor.append!(t)
    // 等待告警
    timer(500) { /* 等待引擎处理 */ }
    // 查询告警并计算延迟
    latestAlert = select top 1 * from alerts where deviceId = "device_test"
    if (latestAlert.size() > 0) {
        latency = latestAlert.alertTime[0] - writeTs
        print("端到端延迟: "   string(latency)   " ms")
    }
}
measureLatency()

在单节点社区版上，端到端延迟通常在5-50ms范围内（取决于数据量和硬件配置）。

六、参数调优

6.1 窗口大小与步长的权衡

窗口大小决定检测灵敏度，步长决定检测频率。这里有一组对比实验数据：

// 不同窗口配置的检测效果对比
configs = [
    dict(`windowSize`step, [1000, 1000]),   // 1秒窗口，1秒步长
    dict(`windowSize`step, [5000, 1000]),   // 5秒窗口，1秒步长
    dict(`windowSize`step, [10000, 5000]),  // 10秒窗口，5秒步长
    dict(`windowSize`step, [30000, 10000])  // 30秒窗口，10秒步长
]
// 用历史数据回测不同配置的检测率和误报率
def backtestConfig(config, data) {
    windowSize = config[`windowSize]
    step = config[`step]
    // 模拟窗口聚合
    result = select count(*) as alertCount,
        sum(iif(deviceId == "device_3", 1, 0)) as truePositive,
        sum(iif(deviceId != "device_3", 1, 0)) as falsePositive
    from data where metricType = `vibration and value > 4.0 group by bar(timestamp, step)
    return result
}

实测结论（50台设备，10分钟数据）：

6.2 分组键（keyColumn）的选择

异常检测引擎按keyColumn分组独立检测。分组键的选择直接影响检测粒度和性能：

• 按deviceId分组：每台设备独立检测，互不影响。适合设备间差异大的场景
• 按deviceId metricType分组：每个设备的每个指标独立检测。适合不同指标有不同阈值要求的场景
• 不设分组键：所有数据混合检测。适合只需检测全局异常的场景

// 按 deviceId   metricType 双键分组
// 需要先创建一个复合键列
def addCompositeKey(t) {
    return select deviceId   "_"   metricType as compositeKey, * from t
}
// 在引擎中使用 compositeKey 作为 keyColumn

6.3 内存占用监控

// 查看流计算引擎的状态和内存占用
getStreamStat().subWorkers
// 查看各引擎的处理速率和积压情况
select name, queuedMsgs, processedMsgs, failedMsgs, lastMsgTime
from getStreamStat().subWorkers
where name like "%tector%"

50台设备、每秒100条记录的场景下，两个引擎的总内存占用约30-50MB，CPU占用率低于5%。

七、从检测到告警的完整闭环

7.1 告警去重与抑制

实际场景中，同一个异常会持续触发告警，需要做去重和抑制：

// 创建告警去重引擎：同一设备同一类型的告警，60秒内只报一次
try { dropStreamEngine("alertDedup") } catch(ex) {}
try { undef(`dedupedAlerts, SHARED) } catch(ex) {}
share streamTable(10000:0, `alertTime`deviceId`alertType`alertDetail`severity`,
    [TIMESTAMP, SYMBOL, SYMBOL, STRING, SYMBOL]) as dedupedAlerts

alertDedup = createSessionWindowEngine(name = "alertDedup",
    sessionGap = 60000, // 60秒会话间隔
    metrics = [, , , , ],
    dummyTable = alerts,
    outputTable = dedupedAlerts,
    timeColumn = `alertTime,
    keyColumn = `deviceId)
subscribeTable(tableName="alerts", actionName="dedupAlerts", offset=0, handler=append!{alertDedup}, msgAsTable=true)

7.2 对接消息中间件

DolphinDB支持通过插件将告警推送到Kafka、MQTT、ZMQ等消息中间件，方便与企业现有的告警通知系统对接。以MQTT为例：

// 加载 MQTT 插件（需提前安装）
loadPlugin("/path/to/plugins/mqtt/PluginMQTTClient.txt")
// 连接 MQTT Broker
conn = mqtt::connect("tcp://broker.example.com:1883", "dolphindb_alert_publisher", "")
// 将告警数据发布到 MQTT Topic
def publishAlert(msg) {
    // 将告警转为 JSON 格式
    jsonStr = toJSON(msg)
    mqtt::publish(conn, "factory/alerts/vibration", jsonStr)
}
subscribeTable(tableName="dedupedAlerts", actionName="publishMqtt", offset=0, handler=publishAlert, msgAsTable=true)

八、性能压测

8.1 吞吐量测试

模拟更高并发场景，测试引擎的处理上限：

def throughputTest(deviceCount, batchSize, iterations) {
    devices = "device_"   string(1..deviceCount)
    totalRows = 0
    startTs = now()
    for (iter in 1..iterations) {
        t = table(batchSize * deviceCount * 2:0,
            `deviceId`timestamp`metricType`value,
            [SYMBOL, TIMESTAMP, SYMBOL, DOUBLE])
        ts = now()
        for (i in 1..batchSize) {
            for (dev in devices) {
                insert into t values(dev, timestamp(ts   i*1000), `temperature, 50   norm(0,2))
                insert into t values(dev, timestamp(ts   i*1000), `vibration, 1.0   norm(0,0.3))
            }
        }
        streamSensor.append!(t)
        totalRows  = t.size()
    }
    elapsed = now() - startTs
    throughput = double(totalRows) / (elapsed / 1000.0)
    print("总写入: "   string(totalRows)   " 行")
    print("总耗时: "   string(elapsed)   " ms")
    print("吞吐量: "   string(throughput)   " 行/秒")
}
// 100台设备，每批100个时间点，跑50批
throughputTest(100, 100, 50)

8.2 压测结果参考

单节点在2000点/秒的写入压力下，端到端检测延迟（P99）仍保持在50ms以内，资源占用可控。如果需要支撑更高的并发，可以通过增加集群节点水平扩展。

九、踩坑记录

9.1 时序数据乱序问题

实际场景中，传感器数据可能因为网络延迟而乱序到达。异常检测引擎默认按数据时间戳排序，如果乱序数据的时间戳早于当前窗口的起始时间，会被丢弃。

解决方案：设置合理的step参数。如果网络延迟通常在2秒以内，将步长设为3-5秒，为乱序数据留出缓冲空间。

9.2 空值处理

传感器数据中可能存在缺失值（设备离线、网络故障等）。窗口聚合函数在遇到空值时行为不同：

• a vg, std等函数会自动跳过空值
• ma vg, mstd等滑动窗口函数在窗口内全为空值时返回空值

建议：在数据写入流表之前，用ffill（前向填充）或interpolate（线性插值）处理缺失值：

// 在订阅回调中预处理数据
def preprocessAndAppend(msg) {
    cleaned = select * from msg context by deviceId, metricType csort timestamp
    update set value = ffill(value)
    streamSensor.append!(cleaned)
}

9.3 引擎重复创建

调试过程中反复执行脚本，如果忘记先dropStreamEngine，会报“引擎已存在”的错误。建议在脚本开头统一清理：

// 清理所有测试引擎和流表
engines = getStreamStat().subWorkers[`name]
for (e in engines) {
    try { dropStreamEngine(e) } catch(ex) {}
}
try { undef(`streamSensor, SHARED) } catch(ex) {}
try { undef(`alerts, SHARED) } catch(ex) {}
try { undef(`aggMetrics, SHARED) } catch(ex) {}
try { undef(`dedupedAlerts, SHARED) } catch(ex) {}

整个过程从数据写入到告警输出在一个系统内闭环完成，无需额外的消息队列或流计算引擎。流批一体的设计让规则验证和上线部署使用同一套代码，开发到上线的周期可以从“周”缩短到“小时”。