如何配置Logstash解析嵌套JSON并将每个嵌套对象转为独立事件？

我来帮你搞定这个Logstash配置问题！针对你给出的JSON数据结构，咱们需要做几个关键处理：把嵌套的时间序列拆成独立文档、清理冗余的字段前缀、保留元数据并正确映射字段类型。下面是完整的配置方案和分步解释：

完整Logstash配置示例 #

input {
  http_poller {
    urls => {
      # 替换成你的公共API地址
      api_data => "https://your-public-api-url.com/data"
    }
    # 拉取间隔，这里设为1小时，按需调整
    schedule => { cron => "0 * * * *" }
    codec => "json"
    # 如果API需要认证，这里可以加headers，比如：
    # headers => {
    #   "Authorization" => "Bearer YOUR_TOKEN"
    # }
  }
}

filter {
  # 第一步：把Time Series的键值对转成数组，方便后续拆分
  ruby {
    code => '
      time_series = event.get("[Time Series]")
      if time_series
        # 把每个日期和对应数据转成{date: "xxx", data: {...}}的数组
        formatted_series = time_series.map { |date, data| { "date" => date, "metrics" => data } }
        event.set("time_series_array", formatted_series)
        # 移除原始的Time Series字段，避免冗余
        event.remove("[Time Series]")
      end
    '
  }

  # 第二步：拆分数组，每个元素生成一个独立事件
  split {
    field => "time_series_array"
    remove_field => ["time_series_array"]
  }

  # 第三步：提取拆分后的日期和指标数据
  mutate {
    add_field => {
      "@timestamp" => "%{[time_series_array][date]}"
      "source_name" => "%{[Meta Data][2. Name]}"
      "last_refreshed" => "%{[Meta Data][3. Last Refreshed]}"
    }
    # 把metrics里的字段提升到顶层
    rename => { "[time_series_array][metrics]" => "metrics" }
    # 移除原始的Meta Data字段（如果不需要保留完整元数据的话）
    # remove_field => ["Meta Data"]
  }

  # 第四步：清理指标字段名，去掉前缀（比如"1. "）并标准化
  ruby {
    code => '
      metrics = event.get("metrics")
      if metrics
        cleaned_metrics = {}
        metrics.each do |key, value|
          # 去掉前缀的数字和点，替换空格为下划线，转小写
          new_key = key.gsub(/^\d+\.\s/, "").gsub(/\s/, "_").downcase
          # 尝试把数值字符串转成整数/浮点数
          begin
            cleaned_value = Float(value)
            cleaned_value = cleaned_value.to_i if cleaned_value.to_i == cleaned_value
          rescue
            cleaned_value = value
          end
          cleaned_metrics[new_key] = cleaned_value
        end
        event.set("metrics", cleaned_metrics)
      end
    '
  }

  # 可选：把metrics里的字段直接提升到顶层（如果不需要嵌套的话）
  # mutate {
  #   merge => { "metrics" => "%{metrics}" }
  #   remove_field => ["metrics"]
  # }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    # 索引名按日期拆分，比如daily-stats-2024.05.20
    index => "daily-stats-%{+YYYY.MM.dd}"
    # 用日期作为文档ID，避免重复数据
    document_id => "%{source_name}-%{@timestamp}"
  }

  # 同时输出到控制台方便调试
  stdout {
    codec => rubydebug
  }
}

关键步骤解释 #

• Input部分：用http_poller插件定时拉取API数据，支持cron表达式设置拉取频率，还能配置请求头处理认证（如果API需要的话）。
• Ruby转换数据结构：原数据的Time Series是嵌套的键值对，咱们用Ruby代码把它转成数组，这样split插件才能把每个日期条目拆成独立的Elasticsearch文档。
• Split拆分事件：把数组里的每个元素拆成单独的Logstash事件，这样每个日期的指标都会成为Elasticsearch里的一条独立数据。
• 字段提取与标准化：
  • 把日期字段设为@timestamp，这是Elasticsearch的时间字段标准格式。
  • 提取元数据里的名称、最后刷新时间等信息，作为每个事件的附加字段。
  • 清理指标字段名：去掉前缀的1. 、2. ，把空格换成下划线并转小写，同时自动把数值字符串转成数字类型，方便后续聚合分析。
• Output部分：配置Elasticsearch的地址、索引名，用来源名称+日期作为文档ID，避免重复拉取时生成重复数据；同时开启stdout输出方便调试。

注意事项 #

1. 如果API返回的日期格式不是ISO标准，需要在mutate后添加date插件重新解析@timestamp，比如：

   date {
     match => ["@timestamp", "yyyy-MM-dd"]
     timezone => "UTC"
   }
   

2. 如果需要保留完整的Meta Data，可以跳过remove_field => ["Meta Data"]的配置。
3. 调试时可以先把拉取间隔设短一点（比如schedule => { every => "1m" }），确认数据处理正确后再调整为生产环境的频率。

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