Logstash - как я могу разделить массив, используя фильтр разделения без цели?

Question

Я пытаюсь разбить массив JSON на несколько событий. Вот пример ввода:

{"results" : [{"id": "a1", "name": "hello"}, {"id": "a2", "name": "logstash"}]}

Вот мой фильтр и выходной конфиг:

filter {
  split {
    field => "results"
  }
}
stdout { 
  codec => "rubydebug"
}

Это создает 2 события, по одному для каждого из JSON в массиве. И это близко к тому, что я ищу:

{                                              
       "results" => {                          
          "id" => "a1",                        
        "name" => "hello"                      
    },                                         
      "@version" => "1",                       
    "@timestamp" => "2015-05-30T18:33:21.527Z",
          "host" => "laptop",                                      
}                                              
{                                              
       "results" => {                          
          "id" => "a2",                        
        "name" => "logstash"                   
    },                                         
      "@version" => "1",                       
    "@timestamp" => "2015-05-30T18:33:21.527Z",
          "host" => "laptop",                                   
}

Проблема заключается во вложенной части "результатов". "результаты" - это значение по умолчанию для целевого параметра. Есть ли способ использовать фильтр разделения без создания вложенного JSON и получить что-то вроде этого:

{                                                                     
          "id" => "a1",                        
        "name" => "hello"                      
      "@version" => "1",                       
    "@timestamp" => "2015-05-30T18:33:21.527Z",
          "host" => "laptop",                                      
}                                              
{                                              
          "id" => "a2",                        
        "name" => "logstash"                   
      "@version" => "1",                       
    "@timestamp" => "2015-05-30T18:33:21.527Z",
          "host" => "laptop",                                   
}

Цель состоит в том, чтобы передать это в вывод ElasticSearch, где каждое событие является документом с document_id => "id". Любые хорошие решения приветствуются!

Answer 1

Если вы знаете, какими будут все поля (как вам кажется), вы можете просто переименовать поля:

    mutate {
            rename => [
                    "[results][id]", "id",
                    "[results][name]", "name"
            ]
            remove_field => "results"
    }

Если вы не знаете, что все поля были, вы могли бы написать ruby фильтр кода, который сделал event['results'].each... и создал новые поля из подполей результатов.