ElasticSearch に Bulk Insert をする

以前の記事でも記載したが、本業のプロジェクトで ElasticSearch を利用したシステムの構築をしようとしている。
Docker で作ったデモ環境では、 Embulk での bulk insert ができていた。 しかし、 AWS Elasticsearch Service では embulk-output-elasticsearch がサポート外のため、利用できなかった。

github.com

なので、別の手段で bulk insert を実装した。

Python の ElasticSearch Client を使う

コードは Python をベースで利用する。
なので、 ElasticSearch の Client として、

Python Elasticsearch Client — Elasticsearch 6.3.0 documentation

を利用する。 インストールは、

pip install elasticsearch

でインストールできる。
基本的には、インストールする Client のバージョンは、 ElasticSearch インスタンスのメジャーバージョンに合わせて利用するのが推奨されていそう。

pypi.org

なので、例えば ElasticSearch の 5.X.X を利用している場合、インストールする Client は

pip install elasticsearch>=5.0.0,<6.0.0

でインストールすることで、5系の中の最新バージョンの Client をインストールすることができる。

bulk insert

ElasticSearch に bulk insert する場合、 helpers の中の bulk method を利用することで bulk insert が実装できる。
参考として、以下に 1000 件のデータに対して bulk insert するコード例を載せる。

from elasticsearch import Elasticsearch, helpers

es = Elasticsearch(host='localhost', port=9200)

data = []

for i in range(1000):
    doc = {'hoge': 'foobar'}
    data.append(
        {'_index':'hoge-index', '_type':'hoge-index-type', '_source':doc}
    )

helpers.bulk(es, data)

100件などの単位で分割して insert したい場合、 for ループ内で適当な件数で bulk method を呼び出すようにすれば良い。

感想

Embulk で insert するのもとても楽だったが、Elasticsearch の helpers を利用して簡単に bulk insert が実装できた。
Elasticsearch の helpers には他にも便利な method が多くあるのでやりたいことを helpers 内の method で実現できるかを考えたほうが良さそうだなと思った。