如何在ELK中提取满足参数匹配的特定事件序列？

方法1：用Elasticsearch EQL查询（最直接） #

EQL是ELK专门用来匹配事件序列的查询语言，完美适配你的需求。

编写EQL查询语句 #

假设你要匹配的参数是user_id（替换成你实际需要匹配的参数名），查询语句如下：

sequence
  [ where event_type == "event_of_type_1" ]
  [ where event_type == "event_of_type_2" and user_id == prev().user_id ]

• 说明：sequence块用来定义序列规则，第一个括号匹配所有event_of_type_1事件，第二个括号专门抓取紧跟在它后面的event_of_type_2事件，同时要求两个事件的user_id完全一致。如果有多个参数需要匹配，直接用and拼接即可，比如user_id == prev().user_id and order_id == prev().order_id。

在Kibana中执行查询 #

1. 打开Kibana的Discover页面，将搜索模式切换为EQL（搜索框上方的选项卡即可选择）。
2. 选择对应的索引模式，粘贴上述查询语句，执行后就能看到所有符合条件的事件对。
3. 要统计序列总数，直接在结果面板中添加Metric可视化组件，选择Count聚合即可得到总数。

方法2：用脚本聚合做统计分析 #

如果需要对匹配到的序列做更复杂的统计（比如查看不同参数对应的序列数量分布），可以使用Elasticsearch的脚本聚合：

统计序列数量与参数分布 #

在Kibana的Dev Tools中执行以下查询：

{
  "size": 0,
  "aggs": {
    "matched_sequences": {
      "scripted_metric": {
        "init_script": "state.events = []",
        "map_script": "state.events.add([type: doc['event_type'].value, params: doc, timestamp: doc['@timestamp'].value])",
        "combine_script": "def sequences = []; for (int i = 0; i < state.events.size() - 1; i++) { def e1 = state.events[i]; def e2 = state.events[i+1]; if (e1.type == 'event_of_type_1' && e2.type == 'event_of_type_2' && e1.params['user_id'].value == e2.params['user_id'].value) { sequences.add([e1: e1, e2: e2]); } } return sequences",
        "reduce_script": "def allSequences = []; for (s in states) { allSequences.addAll(s); } return allSequences"
      }
    },
    "param_stats": {
      "terms": {
        "field": "user_id"
      },
      "aggs": {
        "sequence_count": {
          "value_count": {
            "field": "@timestamp"
          }
        }
      }
    }
  }
}

• 说明：scripted_metric聚合会遍历所有事件，筛选出符合条件的事件对；param_stats聚合可以统计每个参数值对应的序列数量，帮你快速掌握参数分布情况。

方法3：Logstash预处理标记序列 #

如果想在数据入库前就标记出符合条件的序列，方便后续快速查询和统计，可以在Logstash中添加ruby过滤器：

filter {
  ruby {
    code => "
      # 存储上一个event_of_type_1的参数
      if @previous_event_1.nil?
        @previous_event_1 = {}
      end

      if event.get('event_type') == 'event_of_type_1'
        # 保存当前事件的参数和时间戳
        @previous_event_1['params'] = event.to_hash.select { |k| ['user_id', 'other_param'].include?(k) }
        @previous_event_1['timestamp'] = event.get('@timestamp')
        event.set('is_sequence_start', true)
      elsif event.get('event_type') == 'event_of_type_2' && !@previous_event_1.empty?
        # 检查参数是否匹配
        param_match = true
        @previous_event_1['params'].each do |k, v|
          if event.get(k) != v
            param_match = false
            break
          end
        end
        # 检查时间顺序（确保是紧随的事件，这里设1小时内，可按需调整）
        time_diff = event.get('@timestamp') - @previous_event_1['timestamp']
        if param_match && time_diff > 0 && time_diff < 3600
          event.set('is_valid_sequence', true)
          event.set('sequence_start_params', @previous_event_1['params'])
        end
        # 清空上一个event_of_type_1的记录
        @previous_event_1 = {}
      else
        # 其他事件类型，清空记录
        @previous_event_1 = {}
      end
    "
  }
}

• 说明：这个脚本会跟踪最近的event_of_type_1事件参数，当遇到event_of_type_2时自动检查参数匹配度和时间顺序，符合条件就给事件添加is_valid_sequence: true的标记。之后在Kibana中直接过滤这个标记，就能快速找到目标序列，统计操作也会更高效。

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