如何结合transaction与streamstats查找Splunk事件日志时间间隔

<替换为你的基础搜索语句>
| transaction publicationID
| eval event_times = mvzip(_time, _raw)
| mvexpand event_times
| eval _time = mvindex(event_times, 0), raw_event = mvindex(event_times, 1)
| sort publicationID _time
| streamstats current=f window=1 last(_time) as prev_time by publicationID
| eval time_diff = _time - prev_time
| where time_diff > 3600
| eval time_diff_hours = round(time_diff/3600, 2)
| eval prev_time_str = strftime(prev_time, "%Y-%m-%d %H:%M:%S"), current_time_str = strftime(_time, "%Y-%m-%d %H:%M:%S")
| table publicationID prev_time_str current_time_str time_diff_hours
| rename prev_time_str as "上一事件时间" current_time_str as "当前事件时间" time_diff_hours as "间隔小时数"

步骤解释 #

• 基础搜索：替换成你筛选特定发布事件的初始搜索，比如 index=pub_logs sourcetype=publication_events
• transaction 聚合：按 publicationID 将关联事件聚合成事务，确保同ID的所有日志被归为一组
• 打包与展开事件：通过 mvzip 和 mvexpand 将事务内的多值事件拆分为独立行，保留时间戳和原始内容
• 排序：按 publicationID 分组并按时间升序排列，保证事件顺序正确
• 计算时间间隔：用 streamstats 获取每个事件的上一条事件时间，计算时间差（秒为单位）
• 筛选超1小时记录：过滤出时间差大于3600秒的条目
• 格式转换：将时间戳转为易读的字符串格式，时间差转为小时单位并保留两位小数
• 输出格式化：指定输出字段并重命名为直观的中文名称

输出格式 #

最终输出为表格，包含以下字段：

• publicationID：发布事件的唯一标识
• 上一事件时间：前一条关联日志的时间（格式：YYYY-MM-DD HH:MM:SS）
• 当前事件时间：当前日志的时间（格式：YYYY-MM-DD HH:MM:SS）
• 间隔小时数：两个事件间的时间间隔（保留两位小数）

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