Codec不向elasticsearch发送数据

Question

投入: json

{"userid": 125,"type": "SELL"}
{"userid": 127,"type": "SELL"}

LOGSTASH CONF文件：

input {
kafka {
bootstrap_servers => ""
topics => ["topic1"]
codec => "json"
}
}
output {
amazon_es {
hosts => [""]
region => ""
aws_access_key_id => ''
aws_secret_access_key => ''
index => "indexname"
}

stdout { codec => rubydebug }
}

标准输出：

{
"userid" => 127,
"@version" => "1",
"@timestamp" => 2018-10-18T13:54:37.641Z,
"type" => "SELL"
}

输出结果和我想要的完全一样。但这不会进入elasticsearch的主题。如果我不使用json过滤器，整个json将作为“message”进入ES。

任何帮助都将不胜感激。

Answer 1

我想出了问题。我使用了一个名为输入json文件数据的字段，这导致了这个问题。logstash将它作为_type存储在ES中，新版本的elasticsearch不允许这样做。这就是为什么如果我的类型是‘买入’，它将接受条目，如果它是‘出售’，它将拒绝它。谢谢你的帮助!

Answer 2

你试过用json过滤器吗？

input {
    kafka {
        bootstrap_servers => ""
        topics => ["topic1"]
    }
}

filter {
    json {
        source => "message"
    }
}
output {
    amazon_es {
        hosts => [""]
        region => ""
        aws_access_key_id => ''
        aws_secret_access_key => ''
        index => "indexname",
        custom_headers => {'Content-Type': 'application/json'}
    }

    stdout { codec => rubydebug }
}