Mamol27 @Mamol27

Как распарсить json в logstash?

Здравствуйте.
Я беру данные из таблицы из базы с помощью JDBC Source Connector в Kafka топик и забираю их логсташем.
данные получаются следующего вида :

spoiler

"{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"int64\",\"optional\":false,\"field\":\"FIELD_1\"},{\"type\":\"int64\",\"optional\":false,\"field\":\"FIELD_2\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"FIELD_3\"}],\"optional\":false,\"name\":\"TABLE\"},\"payload\":{\"FIELD_1\":20,\"FIELD_2\":4,\"FIELD_3\":52}}"

С помощью json фильтра в логсташе я получаю

only json

{
    "@timestamp" => 2021-09-06T09:34:45.914Z,
       "message" => {
        "payload" => {
            "FIELD_2" => 2,
            "FIELD_1" => 26,
            "FIELD_3" => 57
        },
         "schema" => {
              "fields" => [
                [0] {
                       "field" => "FIELD_1",
                    "optional" => false,
                        "type" => "int64"
                },
                [1] {
                       "field" => "FIELD_2",
                    "optional" => false,
                        "type" => "int64"
                },
                [2] {
                       "field" => "FIELD_3",
                    "optional" => true,
                        "type" => "int32"
                }
            ],
                "name" => "TABLE",
            "optional" => false,
                "type" => "struct"
        }
    },
      "@version" => "1"
}

То есть значения интересующих меня полей - числовые.

После фильтра

filter

filter {
    json {
        source => "message"
        target => "message"
    }
    mutate {
        add_field => {
            "FIELD_1"  => "%{[message][payload][FIELD_1]}"
            "FIELD_2"  => "%{[message][payload][FIELD_2]}"
            "FIELD_3"  => "%{[message][payload][FIELD_3]}"
        }
        convert => {
            "FIELD_1"  => "integer"
            "FIELD_2"  => "integer"
            "FIELD_3"  => "integer"
        }

        remove_field => [ "message" ]

    }
}

output

{
        "FIELD_3" => "52",
        "FIELD_2" => "4",
    "@timestamp" => 2021-09-06T13:01:58.448Z,
        "FIELD_1" => "20",
      "@version" => "1"
}

Однако я пытаюсь получить структура вида

needed

{
    "@version" => "1",
    "FIELD_1" => 20,
    "FIELD_2" => 4,
    "FIELD_3" => 52
    "@timestamp" => 2021-09-06T09:34:45.914Z
}

Почему-то поля так и остаются текстовыми, как взять значения не через sprintf не понимаю.
Есть ли более элегантный способ распарсить эти данные в нужный мне вид?