Как добавить массив словаря через logsta sh фильтр mutate из csv? - PullRequest
Новая
       50

Как добавить массив словаря через logsta sh фильтр mutate из csv?

0 голосов
/ 01 мая 2020

Я написал файл конфигурации logsta sh для загрузки CSV, CSV имеет несколько сведений о кандидате, мне нужно загружать как массив словаря в индексе kibana, а не как словарь dict с индексом. 

filter {
    csv {
        separator => ","
        skip_header => true
        columns => [LoanID,Applicant_Income1,Occupation1,Time_At_Work1,Date_Of_Join1,Gender,LoanAmount,Marital_Status,Dependents,Education,Self_Employed,Applicant_Income2,Occupation2,Time_At_Work2,Date_Of_Join2,Applicant_Income3,Occupation3,Time_At_Work3,Date_Of_Join3]
    }
    mutate { 
        convert => {
            "Applicant_Income1" => "float"
            "Time_At_Work1" => "float"
            "LoanAmount" => "float"
            "Applicant_Income2" => "float"
            "Time_At_Work2" => "float"
            "Applicant_Income3" => "float"
            "Time_At_Work3" => "float"
            }
        } 
    mutate{
        rename => {
            "Applicant_Income1" => "[Applicant][0][Applicant_Income]"
            "Occupation1" => "[Applicant][0][Occupation]"
            "Time_At_Work1" => "[Applicant][0][Time_At_Work]"
            "Date_Of_Join1" => "[Applicant][0][Date_Of_Join]"
            "Applicant_Income2" => "[Applicant][1][Applicant_Income]"
            "Occupation2" => "[Applicant][1][Occupation]"
            "Time_At_Work2" => "[Applicant][1][Time_At_Work]"
            "Date_Of_Join2" => "[Applicant][1][Date_Of_Join]"
            "Applicant_Income3" => "[Applicant][2][Applicant_Income]"
            "Occupation3" => "[Applicant][2][Occupation]"
            "Time_At_Work3" => "[Applicant][2][Time_At_Work]"
            "Date_Of_Join3" => "[Applicant][2][Date_Of_Join]"
            }
        }   
    date {
        match => [ "Date_Of_Join1", "yyyy-MM-dd'T'HH:mm:ss.SSZZ" ]
        }   
    date {
        match => [ "Date_Of_Join2", "yyyy-MM-dd'T'HH:mm:ss.SSZZ" ]
      } 
    date {
        match => [ "Date_Of_Join3", "yyyy-MM-dd'T'HH:mm:ss.SSZZ" ]      
      }       
    }

Я получил поле «Заявитель» как

enter image description here

Но мне нужно, чтобы поле «Заявитель» было массивом словарей, например

enter image description here

Я пробовал add_field, но не работает 

    mutate{
        add_field => {  "[Applicant][Applicant_Income1]" => "Applicant_Income1",
                    "[Applicant][Occupation1]" => "Occupation1",
                "[Applicant][Time_At_Work1]" => "Time_At_Work1",
                "[Applicant][Date_Of_Join1]" => "Date_Of_Join1"
                        }
        }

1 Ответ

2 голосов
/ 01 мая 2020

Квадратные скобки в Logsta sh Фильтры не ведут себя как элементы / записи массива, как в других языках программирования, например, Java.

[Applicant][0][Applicant_Income]

не подходит синтаксис для установки значения поля Applicant_Income первого элемента (индекс на основе нуля) в массиве Applicant. Вместо этого вы создаете подэлементы 0, 1, 2 под элементом Applicant, как показано на рисунке 1.

Чтобы создать массив объектов, вы должны использовать плагин фильтра ruby (https://www.elastic.co/guide/en/logstash/current/plugins-filters-ruby.html). Поскольку вы можете выполнять произвольный код ruby с этим фильтром, он дает вам больше контроля / свободы: 

filter {
  csv {
    separator => ","
    skip_header => true
    columns => [LoanID,Applicant_Income1,Occupation1,Time_At_Work1,Date_Of_Join1,Gender,LoanAmount,Marital_Status,Dependents,Education,Self_Employed,Applicant_Income2,Occupation2,Time_At_Work2,Date_Of_Join2,Applicant_Income3,Occupation3,Time_At_Work3,Date_Of_Join3]
  }

  mutate { 
    convert => {
      "Applicant_Income1" => "float"
      "Time_At_Work1" => "float"
      "LoanAmount" => "float"
      "Applicant_Income2" => "float"
      "Time_At_Work2" => "float"
      "Applicant_Income3" => "float"
      "Time_At_Work3" => "float"
    }
  } 

  ruby{
    code => '
      event.set("Applicant", 
       [
        {
         "Applicant_Income" => event.get("Applicant_Income1"),
         "Occupation" => event.get("Occupation1"), 
         "Time_At_Work" => event.get("Time_At_Work1"),
         "Date_Of_Join" => event.get("Date_Of_Join1")
        },
        {
           # next object...
        }
       ]
    '
  }

  date {
    match => [ "Date_Of_Join1", "yyyy-MM-dd'T'HH:mm:ss.SSZZ" ]
  } 

  date {
    match => [ "Date_Of_Join2", "yyyy-MM-dd'T'HH:mm:ss.SSZZ" ]
  } 

  date {
    match => [ "Date_Of_Join3", "yyyy-MM-dd'T'HH:mm:ss.SSZZ" ] 
  } 

  mutate{
    remove_field => [
      "Applicant_Income1",
      "Occupation1",
      "Time_At_Work1",
      "Date_Of_Join1",
      "Applicant_Income2",
      "Occupation2",
      "Time_At_Work2",
      "Date_Of_Join2",
      "Applicant_Income3",
      "Occupation3",
      "Time_At_Work3",
      "Date_Of_Join3"
    ]
  } 
}

С помощью event.set вы добавляете поле в документ. Первый аргумент - это имя поля, второй - его значение. В этом случае вы добавляете поле «Кандидаты» в документ с массивом объектов в качестве его значения.

event.get используется для получения значения определенного поля в документе. Вы получаете значение, передавая имя поля методу.

Пожалуйста, обратитесь к этому руководству https://www.elastic.co/guide/en/logstash/current/event-api.html, чтобы получить более подробную информацию об API события.

Я надеюсь, что я мог бы помочь вам.

...