logstash怎么解析多层json

Logstash 解析多层 JSON 的实用指南

在日志处理场景中,JSON 格式因结构化、易解析的特性被广泛应用，但实际日志数据常包含多层嵌套的 JSON 结构（如日志主体嵌套请求详情、错误上下文等），如何让 Logstash 准确解析这些多层嵌套数据，是许多开发者面临的挑战，本文将详细介绍 Logstash 解析多层 JSON 的方法，从基础配置到进阶技巧，帮助你高效处理复杂 JSON 日志。

Logstash 解析多层 JSON 的基础方法

Logstash 提供了多种输入、过滤和输出插件，json 插件（位于 filter 阶段）是解析 JSON 数据的核心工具，对于多层 JSON，关键在于正确配置 source 参数（指定原始 JSON 字段）和 target 参数（指定解析后的目标字段），避免数据覆盖或混乱。

基础配置示例

假设日志数据为单行 JSON，包含两层嵌套结构，

{"log_id":"123","timestamp":"2023-10-01T12:00:00Z","message":{"level":"info","content":"User login successful","details":{"user_id":"u001","ip":"192.168.1.1"}}}

目标：将 message 和 details 字段解析为独立字段，而非嵌套在 message 对象内。

Logstash 配置（multiline_json.conf）：

input {
  # 假设从 Filebeat 读取日志（也可改为 stdin 直接测试）
  beats {
    port => 5044
  }
}
filter {
  # 解析原始 JSON 字段（假设原始 JSON 在 message 字段中）
  json {
    source => "message"
    target => "parsed"  # 将解析结果存入 parsed 字段，避免覆盖原始 message
  }
  # 进一步解析 parsed.message.details 下的嵌套字段
  # 方法1：使用 split + mutate 合并字段
  if [parsed][message][details] {
    mutate {
      add_field => {
        "log.level" => "%{[parsed][message][level]}"
        "log.content" => "%{[parsed][message][content]}"
        "user.id" => "%{[parsed][message][details][user_id]}"
        "user.ip" => "%{[parsed][message][details][ip]}"
      }
      remove_field => ["[parsed][message][details]"]  # 清理中间嵌套字段
    }
  }
  # 方法2：使用 ruby 插件深度解析（更灵活，但性能略低）
  # ruby {
  #   code => "
  #     details = event.get('[parsed][message][details]')
  #     if details.is_a?(Hash)
  #       event.set('user.id', details['user_id'])
  #       event.set('user.ip', details['ip'])
  #     end
  #   "
  # }
}
output {
  stdout { codec => rubydebug }  # 输出到控制台查看结果
  # 输出到 Elasticsearch 或其他目标
  # elasticsearch {
  #   hosts => ["localhost:9200"]
  #   index => "logs-%{+YYYY.MM.dd}"
  # }
}

执行效果： 原始 JSON 中的 message.level、message.content、message.details.user_id 等字段会被拆解为顶级字段（如 log.level、user.id），结构更清晰，便于后续处理和查询。

关键参数说明

• source：指定原始 JSON 字符串所在的字段（如 message、log 等），若日志本身就是完整 JSON 行（如 Filebeat 直接发送的 JSON 日志），可省略 input 阶段的 codec => json，直接在 filter 中用 json 插件解析默认字段（Logstash 会自动将事件作为 JSON 解析）。
• target：指定解析后的 JSON 对象存储的目标字段。建议始终配置 target，避免直接覆盖原始字段（如原始字段可能包含非 JSON 数据，需保留用于回溯）。
• skip_on_invalid_json：若原始数据可能包含非 JSON 内容（如行尾追加的异常文本），设置为 true 可跳过无效数据，避免事件处理失败。

处理多层嵌套的进阶技巧

当 JSON 层级较深（如 3 层以上）或结构动态变化时，基础方法可能难以满足需求，此时可结合其他插件或技巧实现更灵活的解析。

使用 ruby 插件递归解析嵌套字段

对于结构复杂或嵌套层级不固定的 JSON，ruby 插件提供了更强大的自定义解析能力，通过 Ruby 代码遍历 Hash 结构，动态提取字段。

示例：假设日志包含多层嵌套：

{"event":{"type":"request","data":{"header":{"user_agent":"Mozilla/5.0"},"body":{"params":{"username":"test","password":"123456"}}}}}

目标：提取 event.data.header.user_agent、event.data.body.params.username 等字段。

配置片段：

filter {
  # 先通过 json 插件初步解析，存入 target_event
  json {
    source => "message"
    target => "target_event"
  }
  # 使用 ruby 递归提取嵌套字段
  ruby {
    code => "
      def extract_nested(hash, prefix = '')
        hash.each do |key, value|
          new_key = prefix.empty? ? key : '#{prefix}.#{key}'
          if value.is_a?(Hash)
            extract_nested(value, new_key)
          else
            event.set(new_key, value)
          end
        end
      end
      extract_nested(event.get('[target_event]'))
    "
  }
  # 清理中间字段（可选）
  mutate {
    remove_field => ["target_event"]
  }
}

效果：生成扁平化字段，如 event.type、event.data.header.user_agent、event.data.body.params.username，层级关系通过 保留，便于查询。

处理数组嵌套：split + json 组合

若 JSON 中包含数组嵌套（如日志记录多个子事件），需先拆分数组，再逐个解析元素。

示例日志：

{"log_id":"456","events":[{"type":"login","result":"success"},{"type":"logout","result":"failed","reason":"timeout"}]}

目标：将 events 数组拆分为独立事件，每个事件包含 type 和 result（或 reason）。

配置片段：

filter {
  # 1. 解析原始 JSON
  json {
    source => "message"
    target => "parsed"
  }
  # 2. 拆分数组 events，为每个元素创建新事件
  split {
    field => "[parsed][events]"
    # 可添加唯一标识，避免事件混淆
    add_tag => ["split_event"]
  }
  # 3. 提取数组元素的字段
  mutate {
    add_field => {
      "event.type" => "%{[parsed][events][type]}"
      "event.result" => "%{[parsed][events][result]}"
      "event.reason" => "%{[parsed][events][reason]}"  # 仅部分事件有该字段
    }
    remove_field => ["[parsed][events]", "parsed"]  # 清理中间字段
  }
}

效果：原始 1 条日志会拆分为 2 条事件，分别包含 login 和 logout 的详情，数组嵌套被转化为平铺结构。

动态字段映射：date、geoip 等插件联动

解析多层 JSON 时，常需结合其他插件对特定字段进一步处理（如时间解析、地理位置解析），此时需先通过 json 或 ruby 提取目标字段，再调用对应插件。

示例：日志包含嵌套的时间戳和 IP 地址：

{"log_data":{"timestamp":"2023-10-01T12:00:00Z+08:00","client_ip":"8.8.8.8","action":"click"}}

目标：将 timestamp 解析为 Logstash 事件时间，client_ip 归一化为地理位置。

配置片段：

filter {
  # 解析 JSON
  json {
    source => "message"
    target => "log_data"
  }
  # 提取时间戳并设置为事件时间（影响 @timestamp 字段）
  date {
    match => ["[log_data][timestamp]", "ISO8601"]  # 匹配 ISO8601 格式
    target => "@timestamp"
  }
  # 提取 IP 并解析地理位置
  geoip {
    source => "[log_data][client_ip]"
    target => "geoip"
  }
  # 提取 action 字段