Logstash - 如何解析嵌套的 JSON 将每个嵌套的对象变成一个事件？

Question

所以，我正在尝试将 logstash 配置为从公共 API 获取 JSON 数据并插入到 Elasticsearch 中。

数据如下：

{
    "Meta Data": {
        "1. Information": "Daily Aggregation",
        "2. Name": "EXAMPLE",
        "3. Last Refreshed": "2018-04-06"
    },
    "Time Series": {
        "2018-04-06": {
            "1. Value1": "20",
            "2. Value 2": "21",
            "3. Value 3": "20",
            "4. Value 4": "21",
            "5. Value 5": "47"
        },
        "2018-04-05": {
            "1. open": "21",
            "2. high": "21",
            "3. low": "21",
            "4. close": "21",
            "5. volume": "88"
        },
        "2018-04-04": {
            "1. open": "20",
            "2. high": "20",
            "3. low": "20",
            "4. close": "20",
            "5. volume": "58"
        },
        "2018-04-03": {
            "1. Value1": "20",
            "2. Value 2": "21",
            "3. Value 3": "20",
            "4. Value 4": "21",
            "5. Value 5": "47"
        },
        ...
    }
}

我不关心元数据，我希望“时间序列”中的每个对象都成为发送到 Elasticsearch 的不同事件。我只是不知道该怎么做。

到目前为止，我刚刚输入配置正确...

input {
  http_poller {
    urls => {
        test1 => "https://www.public-facing-api.com/query?function=TIME_SERIES_DAILY&name=EXAMPLE"
        #headers => {
        #   Accept => "application/json"
        #}
    }
    request_timeout => 60
    # Supports "cron", "every", "at" and "in" schedules by rufus scheduler
    schedule => { cron => "* * * * * * UTC"}
    codec => "json"
  }
}

filter {
    json {
        source => "message"
        target => "parsedMain"
    }
    json {
        source => "[parsedMain][Time Series]"
        target => "parsedContent"
    }
}

output {
  stdout { codec => rubydebug }
}

但它只是将所有内容打印为单个对象。

我还想捕获日期，即每个嵌套对象的名称，并将其设置为 ES 时间戳。此外，id 为 %{date}_%{name}。

有人知道怎么做吗？

Answer 1

为此，您需要一个 ruby​​ 过滤器 + 一个拆分过滤器。需要将Time Series哈希转成数组，然后在数组上拆分：

filter {
    json {
        source => "message"
    }
    ruby {
        code => '
            arrayOfEvents = Array.new()
            ts = event.get("Time Series")
            ts.each do |date,data|
                data["date"]=date # set the date on the sub-object, since we likely need that
                arrayOfEvents.push(data)
            end
            event.set("event",arrayOfEvents)
        '
        remove_field => ["Time Series","Meta Data" ]
    }
    split {
        field => 'event'
    }
}
output {
    stdout { codec => rubydebug }
}

示例输出：

...
{
    "@timestamp" => 2018-04-09T15:01:01.765Z,
      "@version" => "1",
          "host" => "xxx.local",
          "type" => "yyyyy",
         "event" => {
              "date" => "2018-04-03",
         "1. Value1" => "20",
        "5. Value 5" => "47",
        "3. Value 3" => "20",
        "4. Value 4" => "21",
        "2. Value 2" => "21"
    }
}
{
    "@timestamp" => 2018-04-09T15:01:01.765Z,
      "@version" => "1",
          "host" => "xxx.local",
          "type" => "yyyyy",
         "event" => {
           "3. low" => "20",
             "date" => "2018-04-04",
        "5. volume" => "58",
          "1. open" => "20",
          "2. high" => "20",
         "4. close" => "20"
    }
}