工业 IoT 告警治理实战：去重、抖动抑制与群发告警的收敛

// 持续时长确认：越限且连续满 confirmSecs 次采样才告警
@state
def persistentAlarm(value, threshold, confirmSecs) {
    mutable dur = 0
    if (value > threshold) {
        dur += 1
    } else {
        dur = 0
    }
    return iif(dur >= confirmSecs, 1, 0)
}

振动值
  5.5 ┤              ●━━━━━ 告警持续中
  5.0 ┤ ─ ─ ─ ─ 触发阈值（向上穿越才触发）─ ─ ─
  4.5 ┤ ─ ─ ─ ─ 恢复阈值（向下穿越才恢复）─ ─ ─
      ┤    ●──●──●  值在 4.6~5.1 震荡 → 不再反复翻转

// 滞回抑制：双阈值状态机
@state
def hysteresisAlarm(value, onThreshold, offThreshold) {
    mutable alarmed = 0
    if (alarmed == 0 and value > onThreshold) {
        alarmed = 1
    } else if (alarmed == 1 and value < offThreshold) {
        alarmed = 0
    }
    return alarmed
}

// 输入：实时测点流（每台设备每秒一条）
// 输出：带状态翻转标记的告警流
alertOut = table(1:0, `deviceId`ts`value`alarmState, [SYMBOL, DATETIME, DOUBLE, INT])

alertEngine = createReactiveStateEngine(
    name        = "alertHygiene",
    metrics     = <[
        persistentAlarm(value, 5.0, 10) as confirmed,
        hysteresisAlarm(value, 5.0, 4.5) as hysteresis
    ]>,
    dummyTable  = sensorStream,
    outputTable = alertOut,
    keyColumn   = "deviceId"
)

subscribeTable(tableName=`sensorStream, actionName="hygiene",
               handler=tableInsert{alertEngine}, msgAsTable=true)

// 第一级输出（alertOut）带 deviceId，先映射到所属产线
// 假设设备档案（PKEY 表）提供 deviceId → lineId 映射
enriched = select a.deviceId, a.ts, a.confirmed,
                  p.lineId
          from aj(alertOut as a, loadTable("dfs://iot_meta", "deviceProfile") as p, `deviceId)

share enriched as enrichedStream

// 第二级：每 10 秒一个窗口，按产线统计越限比例
groupOut = table(1:0, `lineId`windowEnd`alarmRatio`alarmCount, [SYMBOL, DATETIME, DOUBLE, INT])

convEngine = createTimeSeriesAggregator(
    name        = "groupConverge",
    windowSize  = 10s,
    step        = 10s,
    metrics     = <[sum(confirmed) \ count(confirmed) as alarmRatio,
                   sum(confirmed) as alarmCount]>,
    dummyTable  = enrichedStream,
    outputTable = groupOut,
    timeColumn  = `ts,
    keyColumn   = `lineId
)

subscribeTable(tableName=`enrichedStream, actionName="converge",
               handler=tableInsert{convEngine}, msgAsTable=true)

// 静默配置表：哪台设备、哪个时间窗、静默原因
share table(1:0, `deviceId`startTime`endTime`reason,
            [SYMBOL, DATETIME, DATETIME, SYMBOL]) as silenceCfg

// 发送前过滤：命中静默窗的只记录不通知
toNotify = select a.deviceId, a.ts, a.alarmState,
                  iif(s.deviceId != NULL, 0, 1) as shouldNotify
           from aj(alertOut as a, silenceCfg as a2, `deviceId) ...
           // 简化示意：实际按 deviceId + 当前时间是否落在 [startTime, endTime] 过滤

// 静默配置由排程系统写入，也可用 scheduleJob 定时同步 CMMS 的检修计划

// 每 30 秒扫一次未确认的严重告警，超 5 分钟未确认 → 推升级流
def escalateCheck() {
    pending = select ticketId, deviceId, createdAt
              from loadTable("dfs://iot_txn", "alertTicket")
              where status = `SEVERE_UNCONFIRMED
                and createdAt < now() - 5m
    if (pending.size() > 0) {
        loadTable("dfs://iot", "escalationStream").tableInsert(pending)
    }
}
scheduleJob("escalate_scan", "严重告警升级扫描", escalateCheck,
            now() + 30s, now() + 1d, 'S', "30s")

// 每日告警治理指标
dailyMetrics = select
    date(createdAt)                                   as day,
    count(*)                                          as total_alerts,
    // 噪音比：自动恢复（从未被处理）的告警占比 —— 越低越好
    sum(iif(status = `AUTO_RECOVERED, 1, 0)) \ count(*)   as noise_ratio,
    // 可行动率：引发工单的告警占比 —— 越高越好
    sum(iif(hasTicket, 1, 0)) \ count(*)                  as actionable_rate,
    // 平均确认时长（分钟）：告警到被确认的耗时
    avg((confirmedAt - createdAt) \ 60000)                as mtta_min
from loadTable("dfs://iot_txn", "alertTicket")
group by date(createdAt)

告警量上涨 → 看是哪个来源（重复/抖动/群发）
    → 重复多：检查去重边沿逻辑有没有漏
    → 抖动多：加宽滞回缓冲带，或提高持续确认时长
    → 群发多：检查组收敛的分组维度（按产线？按配电回路？）
    → 都不是：规则本身过敏感，调阈值或降级