如何编写Splunk查询筛选100ms内同标识的关联事件？

需求说明 #

需要提取满足以下条件的事件数据：

• 事件对的发生时间间隔≤100ms
• 事件对拥有完全相同的thirdPartyId和hashCode
  核心逻辑是按thirdPartyId与hashCode的组合分组，组内按时间排序后，逐行计算当前事件与上一事件的时间差，筛选出符合时间间隔要求的事件。

正确的Splunk查询语句 #

| makeresults format=csv data="startTS,thirdPartyId,hashCode,accountNumber
2024-04-16 21:53:02.455-04:00,AAAAAAAA,00000001,11111111
2024-04-16 21:53:02.550-04:00,AAAAAAAA,00000001,11112222
2024-04-16 21:53:02.650-04:00,BBBBBBBB,00001230,22222222
2024-04-16 21:53:02.650-04:00,CCCCCCCC,00000002,12121212
2024-04-16 21:53:02.730-04:00,DDDDDDDD,00000005,33333333
2024-04-16 21:53:02.830-04:00,DDDDDDDD,00000005,33334444
2024-04-16 21:53:02.670-04:00,BBBBBBBB,00000002,12121212
2024-04-16 21:53:02.700-04:00,CCCCCCCC,00000002,21212121"
| eval _time = strptime(startTS, "%Y-%m-%d %H:%M:%S.%3Q%z")
| sort 0 thirdPartyId, hashCode, _time
| streamstats current=f window=1 last(_time) as prev_time by thirdPartyId, hashCode
| eval time_diff = _time - prev_time
| where time_diff <= 0.1
| eval prev_startTS = strftime(prev_time, "%Y-%m-%d %H:%M:%S.%3Q%z")
| table startTS, prev_startTS, thirdPartyId, hashCode, accountNumber, time_diff

语句分步解释 #

1. eval _time = strptime(...)：将自定义的startTS字段转换为Splunk原生的_time时间戳，方便后续时间计算
2. sort 0 thirdPartyId, hashCode, _time：按分组字段+时间排序，0表示不限制结果数量
3. streamstats ...：在每个thirdPartyId+hashCode分组内，获取上一行事件的时间戳存入prev_time
4. eval time_diff = _time - prev_time：计算当前事件与上一事件的时间差（单位：秒，0.1秒=100ms）
5. where time_diff <= 0.1：筛选出时间间隔≤100ms的事件
6. eval prev_startTS+table：格式化输出字段，将prev_time转换为可读时间格式，整理展示所需字段

关于错误尝试的说明 #

你之前用bin _time span=100ms+stats count的方式存在两个关键问题：

• 固定时间桶分组会漏掉跨桶的相邻事件（比如一个事件在2.699s，下一个在2.701s，间隔2ms但分属不同桶）
• stats只会统计分组内的事件数量，无法保留原始事件的详细信息（比如accountNumber），也无法定位具体是哪两个事件满足条件

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