如何在Kusto Engine数据中过滤尖峰并适配新基线值？

需求说明 #

• 处理发动机小时数累加值，需完成：
  1. 过滤数据中出现的数量级增减尖峰异常
  2. 自动识别发动机更换后的新基线（规则：连续5个值与当前基线差异显著时切换基线）
• 最终需按时间周期分箱，基于基线分区计算delta（max-min）

已尝试方案的问题 #

• 使用series_outliers过滤尖峰：调优后仍存在尖峰漏判或有效数据误过滤的情况
• 使用scan函数对比时间差与小时数差：无法实现自动更新基线的逻辑

测试数据与实现代码 #

datatable (['time']: datetime, stored_float: real) [
    datetime(2025-01-01T04:40:49.000Z),real(1178.9),
    datetime(2025-01-19T12:28:15.000Z),real(1530),
    datetime(2025-01-19T20:41:31.000Z),real(1536.7),
    datetime(2025-01-20T04:42:55.000Z),real(1543.8),
    datetime(2025-01-20T05:38:55.000Z),real(154), //负向尖峰
    datetime(2025-01-20T12:35:50.000Z),real(1550.3),
    datetime(2025-01-20T20:40:43.000Z),real(1557),
    datetime(2025-01-21T08:54:32.000Z),real(15682), //正向尖峰
    datetime(2025-01-21T12:40:32.000Z),real(1571.8),
    datetime(2025-01-22T10:39:52.000Z),real(15889), //正向尖峰
    datetime(2025-01-22T12:37:06.000Z),real(1590.8),
    datetime(2025-01-22T21:44:36.000Z),real(1594.8),
    datetime(2025-01-23T13:07:22.000Z),real(1607.2),
    datetime(2025-01-24T13:07:22.000Z),real(16007.2), //新基线
    datetime(2025-01-25T13:07:22.000Z),real(16017.2),
    datetime(2025-01-26T13:07:22.000Z),real(16027.2),
    datetime(2025-01-27T13:07:22.000Z),real(16037.2),
    datetime(2025-01-28T13:07:22.000Z),real(16047.2),
    datetime(2025-01-29T13:07:22.000Z),real(16057.2)
]
| order by ['time'] asc
| scan with_match_id=baseline_partition declare(
    baselineTime:datetime, 
    baselineEngine:real, 
    timeDelta:real, 
    engineDelta:real, 
    consecutive_anomalies:int
) with 
(
    // 初始化基线为第一条记录
    step init: true => 
        baselineTime = ['time'], 
        baselineEngine = stored_float, 
        timeDelta = 0.0, 
        engineDelta = 0.0, 
        consecutive_anomalies = 0;
    // 逐行检查异常，更新连续异常计数，满足条件则切换基线
    step update_state: true => 
        timeDelta = (['time'] - init.baselineTime)/1h,
        engineDelta = stored_float - init.baselineEngine,
        consecutive_anomalies = iff(abs(engineDelta) > timeDelta, init.consecutive_anomalies + 1, 0),
        baselineTime = iff(consecutive_anomalies >= 5, ['time'], init.baselineTime),
        baselineEngine = iff(consecutive_anomalies >= 5, stored_float, init.baselineEngine);
)
// 标记异常记录
| extend anomaly = abs(engineDelta) > timeDelta

预期中间结果（过滤异常前） #

• timeDelta：当前时间与基线时间的差值（单位：小时）
• engineDelta：当前发动机小时数与基线小时数的差值
• anomaly：布尔值，标记abs(engineDelta) > timeDelta的异常记录

内容的提问来源于stack exchange，提问作者Daniel Fynes-Clinton