Splunk转Sentinel迁移：Kusto按指定事件合并文件审计日志

针对迁移中需要合并特定文件审计事件的需求——即Account、Computer、file_path、merge_group完全匹配，1秒时间窗内先出现AccessList=1537的事件，后出现AccessList=4424的事件，之前用summarize + make_set_if的方法无法区分事件顺序和严格匹配起始/结束值，会错误合并同类型事件。可以通过Kusto的窗口函数+分区排序来实现精准匹配，以下是两种可行方案：

方案1：直接筛选并合并符合条件的事件对 #

这个方案直接定位到符合顺序要求的事件对，合并对应数据：

SecurityFileEvents
// 先过滤目标AccessList事件，减少计算开销
| where AccessList in ('1537', '4424')
// 按分组字段分区，每组内按时间升序排序
| partition by Account, Computer, file_path, merge_group 
  (
    sort by TimeGenerated asc
    // 获取前一条事件的AccessList和时间戳
    | extend prev_AccessList = prev(AccessList), prev_TimeGenerated = prev(TimeGenerated)
    // 筛选当前为结束事件（4424）、前一条为起始事件（1537）且时间差≤1秒的记录
    | where AccessList == '4424' and prev_AccessList == '1537' and TimeGenerated - prev_TimeGenerated <= 1s
    // 合并前后事件的EventData为数组
    | extend MergedEvents = pack_array(prev(EventData), EventData)
  )
// 按需输出需要的字段
| project Account, Computer, file_path, merge_group, StartTime = prev_TimeGenerated, EndTime = TimeGenerated, MergedEvents

逻辑说明： #

1. 先过滤出仅包含目标AccessList值的事件，缩小处理范围；
2. 用partition by将数据按指定分组列拆分，每组内按时间排序，保证事件顺序正确；
3. 通过prev()函数关联前一条事件的属性，精准判断是否符合“先1537后4424、间隔1秒内”的条件；
4. 用pack_array合并前后事件的核心数据，实现类似Splunk transaction的合并效果。

方案2：标记事件对后合并所有匹配项 #

如果需要保留起始和结束事件的完整信息，可先标记符合条件的事件，再合并：

// 先标记符合条件的起始/结束事件
let marked_events = SecurityFileEvents
| where AccessList in ('1537', '4424')
| partition by Account, Computer, file_path, merge_group 
  (
    sort by TimeGenerated asc
    // 标记起始事件：当前是1537，下一条是4424且时间差≤1秒
    | extend is_start = (AccessList == '1537' and next(AccessList) == '4424' and next(TimeGenerated) - TimeGenerated <= 1s)
    // 标记结束事件：当前是4424，上一条是1537且时间差≤1秒
    | extend is_end = (AccessList == '4424' and prev(AccessList) == '1537' and TimeGenerated - prev(TimeGenerated) <= 1s)
  );
// 合并标记后的事件对
marked_events
| where is_start or is_end
| partition by Account, Computer, file_path, merge_group 
  (
    sort by TimeGenerated asc
    | summarize MergedEvents = make_set(EventData), TimeRange = range(min(TimeGenerated), max(TimeGenerated), 1s) by Account, Computer, file_path, merge_group
    | where array_length(MergedEvents) == 2
  )

逻辑说明： #

1. 先给符合条件的起始/结束事件打上标记；
2. 过滤出标记后的事件，再按分组列合并，确保每组仅包含一对符合条件的事件；
3. 最终输出合并后的事件集合和时间范围。

这两种方案都能严格区分事件顺序和起始/结束值，避免错误合并同类型事件的问题，适配你的迁移需求。

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