Logstash能否对Filebeat采集的日志执行统计分析？

完全可以通过Logstash实现你需要的统计需求，下面是具体的实现思路和配置示例：

一、调整数据流向（两种可选方案） #

方案1：实时采集时直接通过Logstash处理 #

修改Filebeat配置，将输出目标从ES切换为Logstash：

# filebeat.yml 输出配置段
output.logstash:
  hosts: ["你的Logstash地址:5044"]

方案2：处理ES中已有的历史日志 #

让Logstash从ES读取存量日志进行统计，配置输入插件：

# logstash.conf 输入配置段
input {
  elasticsearch {
    hosts => ["你的ES地址:9200"]
    index => "你的Nginx日志索引名"
    # 可添加时间范围筛选，只处理指定时段的数据
    query => '{"query": {"range": {"@timestamp": {"gte": "2024-01-01T00:00:00Z", "lte": "2024-01-02T00:00:00Z"}}}}'
    size => 1000
    scroll => "5m"
  }
}

二、Logstash统计逻辑实现 #

核心依赖geoip插件解析IP区域信息，结合aggregate插件按时段+区域聚合统计。

1. 解析日志与IP区域信息 #

先通过grok提取Nginx日志中的关键字段，再用geoip解析IP对应的区域，同时生成时段分组字段：

filter {
  # 解析Nginx日志（请根据你的实际日志格式调整grok表达式）
  grok {
    match => { "message" => "%{IP:client_ip} - - \[%{HTTPDATE:request_time}\] \"%{WORD:method} %{URIPATH:path} %{DATA:http_version}\" %{NUMBER:status} %{NUMBER:bytes_sent}" }
  }

  # 将日志时间转换为Logstash标准时间格式
  date {
    match => [ "request_time", "dd/MMM/yyyy:HH:mm:ss Z" ]
    target => "@timestamp"
  }

  # 解析客户端IP，获取国家/城市信息
  geoip {
    source => "client_ip"
    target => "geoip"
    fields => ["country_name", "city_name"]
  }

  # 生成时段分组字段（示例为按小时统计，可改为%Y-%m-%d按天，或%Y-%m-%d %H:%M按分钟）
  ruby {
    code => "event.set('time_slot', event.get('@timestamp').strftime('%Y-%m-%d %H:00:00'))"
  }
}

2. 按时段+区域聚合统计 #

用aggregate插件分组统计请求数、总流量等指标：

filter {
  aggregate {
    # 按时段+国家+城市作为分组ID
    task_id => "%{time_slot}_%{geoip.country_name}_%{geoip.city_name}"
    code => "
      # 初始化统计字段
      map['request_count'] ||= 0
      map['total_bytes'] ||= 0
      # 累加统计值
      map['request_count'] += 1
      map['total_bytes'] += event.get('bytes_sent').to_i
      # 保留分组标识字段
      map['time_slot'] = event.get('time_slot')
      map['country'] = event.get('geoip.country_name')
      map['city'] = event.get('geoip.city_name')
    "
    # 分组完成后输出统计事件
    push_previous_map_as_event => true
    # 超时时间，确保同组数据全部被收集
    timeout => 300
  }
}

三、输出统计结果 #

将统计好的数据输出到ES的专用索引，方便后续查询分析：

output {
  elasticsearch {
    hosts => ["你的ES地址:9200"]
    # 按日期创建统计索引
    index => "nginx-traffic-stats-%{+YYYY.MM.dd}"
    # 用分组ID作为文档ID，避免重复统计
    document_id => "%{time_slot}_%{country}_%{city}"
  }
}

四、注意事项 #

• 确保Logstash已安装geoip和aggregate插件（默认已包含，缺失可执行bin/logstash-plugin install 插件名安装）
• 时段粒度可根据需求灵活调整，修改ruby代码中的strftime格式即可
• 处理历史数据时，需合理设置ES的scroll参数，避免数据读取超时
• 实时处理场景下，aggregate的timeout值要适配你的业务流量，确保同组数据能被完整收集

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