Elasticsearch 输出插件是否支持 elasticsearch 的 _update_by_query？ https://www.elastic.co/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html
【问题讨论】：
标签： elasticsearch logstash logstash-configuration
使用标准的 elasticsearch 插件可以获得相同的结果：
input {
elasticsearch {
hosts => "${ES_HOSTS}"
user => "${ES_USER}"
password => "${ES_PWD}"
index => "<your index pattern>"
size => 500
scroll => "5m"
docinfo => true
}
}
filter {
...
}
output {
elasticsearch {
hosts => "${ES_HOSTS}"
user => "${ES_USER}"
password => "${ES_PWD}"
action => "update"
document_id => "%{[@metadata][_id]}"
index => "%{[@metadata][_index]}"
}
}
【讨论】：
elasticsearch 输出插件只能调用_bulk 端点，即使用Bulk API。
如果你想调用Update by Query API，你需要使用http输出插件并自己在事件内部构造查询。如果您解释了您想要实现的目标，我可以用更多细节更新我的答案。
注意：有一个issue 请求此功能，但两年后它仍然开放。
更新
因此，如果您的输入事件是{"cname":"wang", "cage":11}，并且您想通过查询所有带有"cname":"wang" 的文档来设置"cage":11，那么您的查询需要如下所示：
POST your-index/_update_by_query
{
"script": {
"source": "ctx._source.cage = params.cage",
"lang": "painless",
"params": {
"cage": 11
}
},
"query": {
"term": {
"cname": "wang"
}
}
}
所以您的 Logstash 配置应该如下所示（您的输入可能会有所不同，但我使用 stdin 进行测试）：
input {
stdin {
codec => "json"
}
}
filter {
mutate {
add_field => {
"[script][lang]" => "painless"
"[script][source]" => "ctx._source.cage = params.cage"
"[script][params][cage]" => "%{cage}"
"[query][term][cname]" => "%{cname}"
}
remove_field => ["host", "@version", "@timestamp", "cname", "cage"]
}
}
output {
http {
url => "http://localhost:9200/index/doc/_update_by_query"
http_method => "post"
format => "json"
}
}
【讨论】：