使用ELK栈结合Logstash Grok解析含JSON的日志并导入Elasticsearch

我来帮你搞定这个ELK日志提取的问题！针对你这种带前缀的JSON格式日志，我们可以用「Grok拆分+JSON解析」的组合方案，既高效又易维护，下面是具体实现步骤：

先分析你的日志结构：每行是时间戳 + 业务类型前缀（charge:/manage:） + JSON payload，这种结构没必要用纯Grok硬解析JSON，拆分后用专门的JSON过滤器处理更靠谱。

1. Logstash完整配置示例 #

以下是包含输入、过滤、输出的完整配置（比如命名为logstash.conf）：

输入部分（对接Filebeat） #

假设你用Filebeat推送日志到Logstash，输入配置如下：

input {
  beats {
    port => 5044  # 和Filebeat配置中的输出端口保持一致
  }
}

过滤部分（核心逻辑） #

这部分是实现字段提取的关键，分两步完成：

filter {
  # 第一步：用Grok拆分时间戳、业务类型和原始JSON字符串
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{WORD:log_type}:%{GREEDYDATA:json_payload}" }
    # 解析失败时打标签，方便后续排查问题
    tag_on_failure => ["_grokparsefailure"]
  }

  # 第二步：解析JSON字符串，自动提取键值对
  json {
    source => "json_payload"  # 指定要解析的字段
    remove_field => ["json_payload"]  # 解析完成后删除原始JSON字段，节省存储空间
  }

  # 可选：把自定义时间戳替换为ELK标准的@timestamp字段（推荐）
  date {
    match => ["timestamp", "yyyy-MM-dd HH:mm:ss"]
    target => "@timestamp"
    remove_field => ["timestamp"]
  }
}

输出部分（转发到Elasticsearch） #

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]  # 你的ES地址
    index => "business-log-%{+YYYY.MM.dd}"  # 按日期分索引，便于管理和查询
  }

  # 可选：输出到控制台，用于调试验证
  stdout {
    codec => rubydebug
  }
}

2. 关键配置说明 #

• Grok模式拆解：
  • %{TIMESTAMP_ISO8601:timestamp}：匹配标准格式的时间戳，存入timestamp字段
  • %{WORD:log_type}：匹配charge/manage这类业务类型单词，存入log_type字段
  • %{GREEDYDATA:json_payload}：匹配剩余所有内容（即完整JSON字符串），存入json_payload字段
• JSON过滤器：会自动把JSON中的键值对（比如oldbalance、cardid、posid等）转换成独立字段，比用Grok逐个匹配JSON字段高效得多
• Date过滤器：将自定义时间戳替换为ELK栈通用的@timestamp字段，方便后续在Kibana中按时间维度筛选日志

3. 调试小技巧 #

• 启动前先验证配置语法：

  bin/logstash -f logstash.conf --config.test_and_exit
  

• 如果日志出现_grokparsefailure标签，说明Grok匹配失败，可以复制日志行到Logstash自带的Grok调试工具（或本地调试脚本）中测试模式
• 开启stdout输出后，可在控制台实时查看解析后的字段结构，确认提取是否正确

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