如何在ADX物化视图中使用Scan算子实现异步流程周期检测？

我有多个独立协作的系统，按预定义流程执行任务。这些系统的状态消息存储在ADX的状态消息表中，消息存在不同步和抖动情况。需要识别的流程周期为：

• 系统1从offline切换为online
• 系统2从offline切换为online
• 系统2从online切换回offline
• 系统1从online切换回offline

目前已通过scan算子实现流程检测，但希望将该逻辑迁移到物化视图中以实现异步计算，然而ADX物化视图不支持序列化数据（scan算子依赖序列化状态）。需要解决两个问题：

• 是否可通过行顺序策略等方式在物化视图中使用scan算子？
• 若不可行，请提供替代解决方案。

当前使用的scan算子代码片段：

scan with_match_id=match_id declare (phase: int, session_start: datetime , session_end: datetime) with
(
    step s1:
        system == "S1" and status == "offline"
        => phase = 1;
    step s2:
        system == "S2" and status == "offline"
        => phase = 2;
    step s3:
        system == "S1" and status == "online"
        => phase = 3;
    step s4:
        system == "S2" and status == "online"
        => phase = 4, session_start = iff(isnull(s4.session_start), timestamp, s4.session_start);
    step s5:
        system == "S1" and status == "offline"
        => phase = 5;
    step s6:
        system == "S2" and status == "offline"
        => phase = 6, session_end = iff(isnull(s6.session_end), timestamp, s6.session_end);
)
| project timestamp, match_id, session_start, session_end, phase
| where phase in (4, 6) 
| summarize take_any(session_start), take_any(session_end) by match_id

状态消息表示例：

物化视图中使用scan算子的可行性 #

ADX物化视图不支持scan算子，原因是scan属于状态化算子，依赖序列化的中间状态跟踪流程阶段，而物化视图的设计限制了这类需要维护跨批次全局状态的算子使用。行顺序策略仅能保证数据处理顺序，无法绕过状态序列化的兼容性问题。

替代解决方案建议 #

方案1：更新策略+中间表 #

通过更新策略模拟物化视图的异步计算效果：

1. 创建中间结果表存储识别出的流程周期数据
2. 为原状态消息表添加更新策略，触发时执行scan检测逻辑并将结果写入中间表
3. 查询时直接从中间表读取结果，无需重复计算

示例代码：

// 创建中间结果表
.create table ProcessCycles (match_id: string, session_start: datetime, session_end: datetime)

// 添加更新策略
.alter table StatusMessages policy update 
@'[{"IsEnabled": true, "Source": "StatusMessages", "Query": "StatusMessages | scan with_match_id=match_id declare (phase: int, session_start: datetime , session_end: datetime) with (step s1: system == \"S1\" and status == \"offline\" => phase = 1; step s2: system == \"S2\" and status == \"offline\" => phase = 2; step s3: system == \"S1\" and status == \"online\" => phase = 3; step s4: system == \"S2\" and status == \"online\" => phase = 4, session_start = iff(isnull(s4.session_start), timestamp, s4.session_start); step s5: system == \"S1\" and status == \"offline\" => phase = 5; step s6: system == \"S2\" and status == \"offline\" => phase = 6, session_end = iff(isnull(s6.session_end), timestamp, s6.session_end);) | project match_id, session_start, session_end, phase | where phase == 6 | summarize take_any(session_start), take_any(session_end) by match_id", "IsTransactional": false, "PropagateIngestionProperties": false}]'

方案2：窗口函数+表连接替代scan #

将流程拆分为独立状态转换事件，通过窗口函数和连接匹配完整流程：

1. 分别提取S1、S2的关键状态转换事件（offline→online、online→offline）
2. 基于时间顺序匹配符合流程逻辑的事件序列
3. 聚合得到完整流程周期

示例代码：

// 提取S1的状态转换事件
let s1_events = StatusMessages
| where system == "S1"
| order by timestamp asc
| extend prev_status = prev(status)
| where prev_status != status
| project timestamp, transition = strcat(prev_status, "→", status);

// 提取S2的状态转换事件
let s2_events = StatusMessages
| where system == "S2"
| order by timestamp asc
| extend prev_status = prev(status)
| where prev_status != status
| project timestamp, transition = strcat(prev_status, "→", status);

// 匹配完整流程序列
s1_events
| where transition == "offline→online"
| join kind=inner (s2_events | where transition == "offline→online") on $left.timestamp < $right.timestamp
| join kind=inner (s2_events | where transition == "online→offline") on $right.timestamp < $right1.timestamp
| join kind=inner (s1_events | where transition == "online→offline") on $right1.timestamp < $right2.timestamp
| project session_start = $left.timestamp, session_end = $right2.timestamp
| summarize take_any(session_start), take_any(session_end) by bin(session_start, 1h) // 可根据实际情况调整分组逻辑

方案3：函数+周期性查询 #

将scan逻辑封装为函数，通过Azure Logic Apps或Azure Functions等定时执行函数，将结果写入目标表，实现异步计算效果。

内容的提问来源于stack exchange，提问作者Rene