Zenn
Open8

elasticsearchの公式ドキュメントData streamsを読む

Elasticsearch
bookstorebookstore

data stream とは？

ログ、イベント、メトリクスといった継続的に生成される時系列データを保存＆検索する仕組みっぽい。

時系列データは複数のインデックスによって管理される。インデックスのプレフィックスに .ds- とつけられるらしい。data stream = 時系列データを扱うために複数のインデックスをまとめて管理する仕組みと理解してもよさそう。

次の設定が必要になる。

  • 同じ index template。おそらく、data streamで管理されるインデックスには同じindex templateが使われるのだと思う。
  • タイムスタンプ。
bookstorebookstore

検索と書き込みはどうなるの？

データ操作の基本である検索と書き込みはどうなるのだろうか？

検索は、data streamが管理する複数のインデックスにまとめて行うことができる。通常、これらのインデックスを意識する必要はなさそう。

書き込みは最も新しいインデックスに対してのみ行われる。古いインデックスに対して直接ドキュメントを追加しようとしても拒否される。基本的に追加のみのユースに適しているらしい。ただ、保存したドキュメントの更新と削除が絶対にできないというわけでもなく、クエリAPIによってドキュメントの変更はできるそう。

Update documents in a data stream by query | Elasticsearch Guide [8.1]

bookstorebookstore

data stream を作るには？

通常、次のステップで作成するとのこと。

  • create index lifecycle policy
  • create component template
  • create index template
  • create data stream
  • secure data stream
bookstorebookstore

rollover とは？

インデックスエイリアスに対し、インデックスの付け替えを行う操作。付け替え先のインデックスが存在しなければ、作成も行うみたいです。

Rollover API | Elasticsearch Guide[8.1]

POST /<rollover対象のインデックスエイリアス>/_rollover/

一連の操作の例です。

PUT log-000001

// エイリアスの作成
PUT log-000001/_alias/log-alias

// rolloverの実行
// ボディに条件を指定できる
POST log-alias/_rollover
{
  "conditions": {
    "max_docs": 10000
  }
}

上記の例では付け替え先のインデックスを自動的に作成するリクエストになります。この場合、後ろの 000001 がインクリメントされたインデックス名が使われます。つまり、 log-000002 インデックスが作成されるということになります。付け替え先のインデックスを自動的に作成してもらうためにはもともとのインデックスの最後に - と数字がついている必要があります。それ以外のパターンになっているインデックス名の場合、自動的にインデックスを作成することができず、リクエストが拒否されます。

increment index names for alias | Elasticsearch Guide [8.1]

また、自動的に作成されるインデックスは古いインデックスの設定を引き継ぎません。設定するにはrolloverのリクエストで明示的に指定するかインデックステンプレートを事前に作成します。

specify settings during rollover | Elasticsearch Guide [8.1]

rollover を用いた運用

rolloverを使うと、簡単なインデックスの運用ができそうだったので試してみました。

イメージはこんな感じになります。

リクエストはこんな感じになります。

// インデックステンプレートを事前に用意します。
// log- から始まるインデックスは log-search-alias エイリアスがつくようにしました。
PUT _index_template/log-index-template
{
  "index_patterns": ["log-*"],
  "template": {
    "aliases": {
      "log-search-alias": {}
    }
  }
}

// インデックスを作成し別のエイリアスを付けます。
PUT log-000001
PUT log-000001/_alias/log-alias

// 一件ドキュメントを保存します。
PUT log-alias/_doc/1
{
  "name": "bookstore-1"
}

// ロールオーバーを行います。
POST log-alias/_rollover

// 別のドキュメントを保存します。
PUT log-alias/_doc/1
{
  "name": "bookstore-2"
}

// bookstore-2 のみ検索結果に表れます。
GET log-alias/_search
{
  "query": {
    "match_all": {}
  }
}

// bookstore-1, bookstore-2 が検索結果に表れます。
GET log-search-alias/_search
{
  "query": {
    "match_all": {}
  }
}

log-search-alias を検索用エイリアス、 log-alias を書き込み用エイリアスとすることができます。rolloverするたびにインデックスは増えていきますが、インデックステンプレートで自動的に検索用エイリアスが付与する仕組みにしています。また、書き込み用エイリアスはrolloverの仕組みによって常に新しいインデックスを指すようになります。rolloverのconditionでドキュメント数や容量を指定すれば簡単な運用としては事足りるのかもしれませんね。

bookstorebookstore

shrink とは

shrink とは、現在あるインデックスのプライマリシャード数を少なくしつつ、別インデックスに移行する操作のことみたいです。

Shrink index API | Elasticsearch Guide

shrink を行う上での制約は大きく二つあります。

  • 縮小先のシャード数は元の数の倍数であること。例えば８のシャード数を持つインデックスの場合、４、２、１のいずれか。
  • 実行する前に全てのプライマリシャードを一つのノードに移動させること。
  • 実行する前にインデックスが read-only になっていること。
  • インデックスの health status が green であること。

縮小元の倍数からのみ選択できる点を鑑みると、インデックスを最初に作成するときに縮小させやすいプライマリシャード数にしておいた方が良さそうですね。例えば、11にしてしまうと1にしかshrink出来ません。

reindexをすれば好きな数に縮小できますが、操作が若干複雑になります。また、reindexの方がCPUのパワーを必要とする重い作業であるという情報もあります。基本的にプライマリシャードを縮小することが事前にわかっているのであれば、shrinkを使うことを前提に設定を施す方がよさそうですね。

https://discuss.elastic.co/t/what-the-difference-between-shrink-api-and-reindex-api/228063

必要条件ではありませんが、実行前にレプリカシャードを無くしておくと、新しいインデックスのシャードの割り当てが簡単になるのでやっておくといいそうです。

shrink するインデックスを作成し、データを登録

プライマリシャード数６、レプリカシャード数２のインデックス source-index を作成します。

source-index
PUT source-index
{
  "settings": {
    "number_of_shards": 6,
    "number_of_replicas": 2
  }
}

ドキュメントも適当に登録してみました。

登録したドキュメント
   "hits" : [
      {
        "_index" : "source-index",
        "_id" : "w36yMYABDznQvTjsAOtY",
        "_score" : 1.0,
        "_source" : {
          "type" : "run",
          "distance" : "5000",
          "date" : "2022-04-16"
        }
      },
      {
        "_index" : "source-index",
        "_id" : "xH6yMYABDznQvTjsGOuu",
        "_score" : 1.0,
        "_source" : {
          "type" : "walk",
          "distance" : "1000",
          "date" : "2022-04-18"
        }
      },
      {
        "_index" : "source-index",
        "_id" : "wn6xMYABDznQvTjs9OuP",
        "_score" : 1.0,
        "_source" : {
          "type" : "walk",
          "distance" : "2000",
          "date" : "2022-04-15"
        }
      }
    ]

実行前の準備

次に実行前の条件を揃えます。公式リファレンスに「こうするといいよ」っていうAPI呼び出し例があるのでそのまま使いましょう。

PUT source-index/_settings
{
  "settings": {
    "index.number_of_replicas": 0,                                
    "index.routing.allocation.require._name": "28c8e3bf06a7",
    "index.blocks.write": true                                    
  }
}

リクエストボディのそれぞれの意味は次の通りです。

  • index.number_of_replicas でレプリカシャードの数を変更します。
  • index.routing.allocation.require._name でプライマリシャードを集めるノードを指定します。
  • index.blocks.write でこのインデックスに対する書き込みをブロックします。

インデックスレベルでのシャード割り当てに関するフィルタリングの設定は次のドキュメントで詳しく解説されているようです。この内容だけでも実験し甲斐がある内容です。今回は単純にshrinkを行うインデックスのシャードを一つのノードに集めるために設定します。

Index-level shard allocation filtering | Elasticsearch

shrink の実行

ではでは、shrinkを実行してプライマリシャード数を変更します。今回は２にしてみようと思います。

shrinkの実行
PUT source-index/_shrink/shrinked-index
{
  "settings": {
    "index.number_of_shards": 2,
    "index.number_of_replicas": 1,
    "index.routing.allocation.require._name": null,
    "index.blocks.write": null
  }
}

実行後、 shrinked-index の設定値を確認してみます。正しくプライマリシャード数が縮小されているのが分かります。

GET shrinked-index

// response 一部抜粋
"number_of_shards" : "2",

登録してあったドキュメントを検索してみます。こちらも問題なく取得できます。

GET shrinked-index/_search
{
  "query": {
    "match_all": {}
  }
}

// response 一部抜粋
     {
        "_index" : "shrinked-index",
        "_id" : "w36yMYABDznQvTjsAOtY",
        "_score" : 1.0,
        "_source" : {
          "type" : "run",
          "distance" : "5000",
          "date" : "2022-04-16"
        }
      },

元々のインデックスの設定を戻すには次のようにします。

PUT source-index/_settings
{
  "settings": {
    "index.number_of_replicas": 2,
    "index.routing.allocation.require._name": null,
    "index.blocks.write": null
  }
}
bookstorebookstore

Force merge とは？

Lucene インデックスのマージを行い、削除マークがついたドキュメントを削除し、インデックスの容量を小さくする操作のことみたいです。

Force Merge | Elasticsearch

Elasticsearchにおけるシャードの実態はLuceneインデックスと言われるものに該当します。Luceneインデックスは内部にセグメントと呼ばれるデータの塊を持っていて、ドキュメントはそこで管理されています。イメージを描いてみました。

Luceneインデックスはイミュータブルであり、一度作成すると変更はできません。そのため、Luceneを利用するElasticsearchでは、ドキュメントの作成・変更・削除という操作をセグメントに対して書き込んでいくことになります。この時、セグメントが一定の数になると複数のセグメントを一つのセグメントに合体させる操作がLuceneによって行われます。これがマージです。マージをすると、削除マークが付けられていたドキュメントが実際のデータから削除されるためデータ容量が減るというわけですね。