ElasticSearchで、documentをupsertする

## TL; DR
ElasticSearchは、以下のような特徴があります。

* indexとdocument idの組み合わせてデータを管理し、不要なデータはindex単位で削除していく
    * 公式で使用されている通り、時系列データを1日単位でindexを作る
    * 古い不要なindexを1日単位で削除できるようにする
    * index単位の処理は高速
* document idもElasticSearchに任せ、こちらで指定しないほうが性能的に良い

なので、つまり

  * document idをユーザ側で指定することは非推奨
    * どこかに書いてあったはずですが失念しました。。。
  * その指定したdocument idのdocumentをupdateすることは非推奨
  * 当然、upsertも非推奨

ということになります。
でも、必要に迫らせてやらざるを得ない場合もあります。
これをどうやるか？というお話です。

## 確認環境

docker-composeで環境を準備します。
kibanaのDev toolsで確認することにします。

```yml
version: '3.9'

services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.15.1
    environment:
      - discovery.type=single-node
      - cluster.name=docker-cluster
      - node.master=true
      - node.data=true
      - xpack.security.enabled=false
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms1024m -Xmx1024m"
    volumes:
      - ./elasticsearch:/usr/share/elasticsearch/data
    ulimits:
      memlock:
        soft: -1
        hard: -1
    networks:
      - esnet

  kibana:
    image: docker.elastic.co/kibana/kibana:7.15.1
    ports:
      - 5601:5601
    depends_on:
      - elasticsearch
    environment:
      ELASTICSEARCH_HOSTS: http://elasticsearch:9200
    networks:
      - esnet

volumes:
  esdata:
    driver: local

networks:
  esnet:
```

Dev toolsは、ここから使えます。
![devtools.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109411/c1e2f886-fa31-9f0f-03da-f7a8f70e84a8.jpeg)


## upsertとは（やりたいこと）

* documentが存在しない場合は、insertする
* documentが存在する場合は、updateする
    * ただし全置換ではない
    * 更新対象のkey:valueだけをPOSTしたら、そのkeyに対応するvalueだけがupdateされる
    * 存在しないkeyの場合は追加される

## insert（指定したidのdocumentが存在しない場合）

documentは、bulk APIでrequestするものとします。

[公式のここ](https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html#bulk-api-request-body)を見ると、bulk APIでPOSTする際、request bodyに、 **'_op_type': 'index'** を指定するとよいようです。

実際のbulk APIでの例を示します。

```json
POST _bulk
{"index":{"_index":"index1","_id":"1"}}
{"field":"value"}
```

responseは以下となりました。

```json
{
  "took" : 2434,
  "errors" : false,
  "items" : [
    {
      "index" : {
        "_index" : "index1",
        "_type" : "_doc",
        "_id" : "1",
        "_version" : 1,
        "result" : "created",
        "_shards" : {
          "total" : 2,
          "successful" : 1,
          "failed" : 0
        },
        "_seq_no" : 0,
        "_primary_term" : 1,
        "status" : 201
      }
    }
  ]
}
```

こうすることで、indexが"index1"、idが"1"のdocumentがindexされます。
responseを確認してみましょう。
ちゃんとindexされていました。

```json
# GET index1/_doc/1
{
  "_index" : "index1",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 1,
  "_seq_no" : 0,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "field" : "value"
  }
}
```

ところが、公式を参照すると、こうも書かれています。

> If the document exists, replaces the document and increments the version. The following line must contain the source data to be indexed.

要するに、後勝ちで、documentの全fieldがreplaceされてしまいます。
updateといえばそうかもしれませんが、必要なところだけ更新したいと思った時には、このままではうまく使えません。

事前に対象documentをgetしておいて、必要なところだけ差し替えたdocumentを作り、その後にPOSTしなければいけません。
アプリ側に余計な処理の作りこみが発生してしまいます。。。。
DB側でできる処理は、できるだけDB側に寄せたいです。

## update（指定したidのdocumentが存在する場合）

[公式のここ](https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html#bulk-api-request-body)を見ると、bulk APIでPOSTする際、request bodyに、 **'_op_type': 'update'** を指定するとよいようです。

実際のbulk APIでの例を示します。

```json
# POST _bulk
{ "update" : {"_id" : "1", "_index" : "index1"} }
{"doc" : {"field" : "value2"} }
```

responseは、以下となりました。

```json
{
  "took" : 19,
  "errors" : false,
  "items" : [
    {
      "update" : {
        "_index" : "index1",
        "_type" : "_doc",
        "_id" : "1",
        "_version" : 2,
        "result" : "updated",
        "_shards" : {
          "total" : 2,
          "successful" : 1,
          "failed" : 0
        },
        "_seq_no" : 1,
        "_primary_term" : 1,
        "status" : 200
      }
    }
  ]
}
```

indexされたはずの内容を確認してみましょう。
responseは以下の通りで、key **field** の値がちゃんと **value2** となっていました。


```json
# GET index1/_doc/1
{
  "_index" : "index1",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 2,
  "_seq_no" : 1,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "field" : "value2"
  }
}
```

しかし、これでもやりたかったupsertはできません。
なぜなら、updateの場合は、対象documentがないとエラーになるようです。
つまり、updateするdocument idが存在しない場合、errorとなってしまいます。


## やりたかったupsert（本命）

指定したidのdocumentが存在し、あと勝ちで部分的に更新したい場合です。
こちらが本命です。

結論からすると、[こちらにやり方が書かれていました](https://github.com/elastic/elasticsearch-js/issues/731)。
updateのオプションに、**doc_as_upsert : true** と指定しろと書かれています。
書き方のサンプルが、[公式のここに書かれていました](https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html#bulk-update)。

まず、上記の操作で作成されている既存のindexを削除してしまいます。

```json
# DELETE index1
{
  "acknowledged" : true
}
```

upsertの挙動を確認するために、連続して以下3つのqueryをPOSTします。  

```json
# POST _bulk
{ "update" : {"_id" : "1", "_index" : "index1", "retry_on_conflict" : 3} }
{"doc" : {"field" : "value"}, "doc_as_upsert" : true }

# POST _bulk
{ "update" : {"_id" : "1", "_index" : "index1", "retry_on_conflict" : 3} }
{"doc" : {"field2" : "value2", "field5" : "value5"}, "doc_as_upsert" : true }

# POST _bulk
{ "update" : {"_id" : "1", "_index" : "index1", "retry_on_conflict" : 3} }
{"doc" : {"field2" : "value2-1"}, "doc_as_upsert" : true }
```

上記3つのPOSTの結果を確認してみます。


```json
# GET index1/_doc/1
{
  "_index" : "index1",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 5,
  "_seq_no" : 4,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "field" : "value",     # <- 1. insert: 最初に追加
    "field2" : "value2-1", # <- 3. update: value2 を value2-1 へ更新
    "field5" : "value5"    # <- 2. upsert: keyとvalueを追加
  }
}

```

意図通り、upsertが出来ています。
更新したいkey : valueだけをupdateすることも出来ました。