我们知道 Elasticsearch output plugin 为我们在 Logstash 的 pipeline 中向 Elasticsearch 的写入提供了可能。我们可以使用如下的格式向 Elasticsearch 写入数据：

  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "data-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "NtC7cM-GKQWOxqamHd1R"
    ssl => true
    ca_trusted_fingerprint => "d464eed5d00a20908318b6a1de38f88daf3a867177123def4c34aa2272571aaf"
  }

在向 Elasticsearch 写入数据的时候，目前它有四种操作：

• index：索引文档（来自 Logstash 的事件）。
• delete：通过 id 删除文档（此操作需要一个id）
• create：索引文档，如果索引中已存在该 id 的文档，则失败。
• update：通过 id 更新文档。 Update 有一个特殊情况，你可以 upsert — 更新文档（如果不存在）。 请参阅 doc_as_upsert 选项。 注意：这在 Elasticsearch 1.x 中不起作用且不受支持。 请升级到 ES 2.x 或更高版本以将此功能与 Logstash 一起使用！

一个 sprintf 样式的字符串，用于根据事件的内容更改操作。 值 %{[foo]} 将使用 foo 字段进行操作。 如果 resolved action 不在 [index, delete, create, update] 中，事件将不会发送到 Elasticsearch。 相反，事件将被发送到管道的死信队列 (DLQ)（如果启用），或者将被记录并删除。

在实际的使用中，假如我们的操作不是 index，delete create 或 update 其中的一种，那么我们该怎么办呢？比如我们想根据一定的条件来更新文档，就像 update by query 那样？我们该怎么办呢？

幸运的是，Logstash 提供了一个叫做 HTTP output plugin。它可以帮我解决这个问题。

准备数据

首先，我们来创建如下的一个索引：

PUT customer/_doc/2
{
  "id": 2,
  "timestamp": "2019-08-11T17:55:56Z",
  "paymentType": "Visa",
  "name": "Darby Dacks",
  "gender": "Female",
  "ip_address": "77.72.239.47",
  "purpose": "Shoes",
  "country": "Poland",
  "age": 55,
  "offer": false 
}

我们在 Kibana 中输入上面的命令来创建一个叫做 customer 的索引。它的 id 为 2。

更新数据

接下来，我们需要按照一定的条件来更新我们的数据。比如，我们想把 paymentType 为 Visa，并且年龄大于或等于 55 岁的人的 offer 设置为 true。在 Kibana 中正常的命令是这样的：

POST customer/_update_by_query
{
  "query": {
    "bool": {
      "must": [
        {
          "match": {
            "paymentType.keyword": "Visa"
          }
        },
        {
          "range": {
            "age": {
              "gte": 50
            }
          }
        }
      ]
    }
  },
  "script": {
    "source": "ctx._source.offer = params.offer",
    "lang": "painless",
    "params": {
      "offer": true
    }
  }
}

我们可以对 Logsthash 做如下的 pipeline 设计：

logstash.conf

input {
  generator {
    message => '{"id":2,"timestamp":"2019-08-11T17:55:56Z","paymentType":"Visa","name":"Darby Dacks","gender":"Female","ip_address":"77.72.239.47","purpose":"Shoes","country":"Poland","age":55}'
    count => 1
  }
}
 
filter {
    json {
        source => "message"
    }
 
    if [paymentType] == "Mastercard" {
        drop {}
    }
 
    mutate {
        remove_field => ["message", "@timestamp", "path", "host", "@version", "log", "event"]
    }
 
}
 
output {
  stdout {
    codec => rubydebug
  }
  http {
      url => "https://localhost:9200/customer/_update_by_query"
      user => "elastic"
      password => "Y+6tv9jejPl=W4IGrTD="
      http_method => "post"
      format => "message"
      content_type => "application/json"
      message => '{"query":{"bool":{"must":[{"match":{"paymentType.keyword":"%{paymentType}"}},{"range":{"age":{"gte":"%{age}"}}}]}},"script":{"source":"ctx._source.offer = params.offer","lang":"painless","params":{"offer":true}}}' 
      cacert => "/Users/liuxg/elastic/elasticsearch-8.6.1/config/certs/http_ca.crt"
  }
}

在上面，我们在 message 中通过一个查询，匹配到 paymentType.keyword 为 Visa，并且 age 为大于等于 55 的文档，我们设置该用户为促销对象。把他的 offer 值设置为 true。这个在实际的使用中，依据自己的条件来进行配置。在上面，cacert 为我们的 Elasticsearch 的证书文件位置。具体使用，请参考文档。

我们接下来运行 Logstash 的 pipeline：

./bin/logstash -f logstash.conf

 

在上面我们可以看出来信息的输出。我们在 Kibana 中使用如下的命令来检查更新后的文档：

GET customer/_search

{
  "took": 0,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 1,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": "customer",
        "_id": "2",
        "_score": 1,
        "_source": {
          "offer": true,
          "country": "Poland",
          "gender": "Female",
          "purpose": "Shoes",
          "name": "Darby Dacks",
          "id": 2,
          "ip_address": "77.72.239.47",
          "age": 55,
          "timestamp": "2019-08-11T17:55:56Z",
          "paymentType": "Visa"
        }
      }
    ]
  }
}

很显然，我们上面的 offer 值现在变为 true，而不是之前的 false。