Python で Elasticsearch を操作するなら、公式の elasticsearch.py を入れることになるのは避けられません。最近、Python で Email Log を受け取って Elasticsearch に書き込む処理を作っていて、共有したい知見がかなりたまりました。AWS Lambda は実行時間で課金されるため、Lambda 関数を書くときは秒単位で気を遣います。0.5 秒を超えるたびに頭を抱えていました。そこで、最初は .index で1件ずつ挿入し、次に streaming_bulk を使い、最後はどうにも我慢できず parallel_bulk と streaming_bulk を組み合わせました。ここでは、これらの方法の違いと踏んだ落とし穴についてお話しします！
1件ずつの追加
最初は、Elasticsearch にデータを挿入するとなると、es.index() を思い浮かべてループで回すでしょう。1件だけ挿入するならそれで問題ありません。しかし N 件を挿入する必要があるなら、これは非常によくない方法です！ es.index() を呼ぶたびに、Python はデータを Elasticsearch に1回ずつ送信します。10件あれば10回送信することになります。1件を送信して Elasticsearch の処理完了を待つのに 100ms かかるなら、10件で 1 秒かかります。ですから複数件の場合、es.index() はあまりおすすめできません。サンプルコード
from elasticsearch import Elasticsearch
es = Elasticsearch("http://127.0.0.1:9200/")
es.index(index="test-index", doc_type='tweet', id=1, body={
"name": "Karl",
"email": "karl@gmail.com"
})シングルスレッドのバルク追加
実は Elasticsearch 自体がバルク操作をサポートしています。リスト全体をまとめて Elasticsearch に送り、処理を任せることができます。書き込み、更新、削除のいずれにも使えます。個別に es.index() を呼ぶよりずっと速くなります！ただし、よく見ると、データ量が増えると実行速度はやはり遅くなっていきます。Elasticsearch サーバー自体はまとめて処理できますが、ネットワーク転送にも時間がかかるためです。データ量が増えると、送信は1回で済んでも、その1回のデータが 30MB になることもあり、やはり遅くなってしまいます。それでも私は満足できませんでした。
バルク処理では、各データ（Dict）に _type と _index を設定し、リストにまとめて Elasticsearch に送信します。あとは Elasticsearch が処理してくれます。
サンプルコード
from elasticsearch import Elasticsearch
from elasticsearch import helpers as es_helpers
es = Elasticsearch('http://127.0.0.1:9200')
data = [
{'name': 'Karl 1', 'email': 'karl@gmail.com', '_type': 'user', '_index': 'user_table', '_id': 1},
{'name': 'Karl 2', 'email': 'karl@gmail.com', '_type': 'user', '_index': 'user_table', '_id': 2},
{'name': 'Karl 3', 'email': 'karl@gmail.com', '_type': 'user', '_index': 'user_table', '_id': 3}
]
es_helpers.bulk(es, data)マルチスレッドのバルク追加
最後の決戦は、複数のスレッドで並列に Elasticsearch へデータを送信する方法です。データ量が多いときは、確かにシングルスレッドよりずっと速くなります。しかしデータ量が少ないと、シングルスレッドより遅くなることもわかります。スレッドを作成する処理自体に時間がかかるからです。データが10件しかないなら、元のスレッドで送信すれば 300ms ほどで済むかもしれません。しかし5件ごとにスレッドを割り当てると、600ms かかることもあります。マルチスレッドには小さな落とし穴があります。parallel_bulk を実行しても、実は何も起きません。データは実際には挿入されていません。parallel_bulk はジェネレーターを返すので、そのジェネレーターを一度最後まで回して初めて、データが挿入されるのです。最初は気づかず、そのせいで1時間分のデータを失いました。詳細は以下のサンプルをご覧ください。
参考文献：elasticsearch2.1,paralle_bulk don’t work,why?
サンプルコード
from collections import deque
from elasticsearch import Elasticsearch
from elasticsearch import helpers as es_helpers
es = Elasticsearch('http://127.0.0.1:9200')
data = [
{'name': 'Karl 1', 'email': 'karl@gmail.com', '_type': 'user', '_index': 'user_table', '_id': 1},
{'name': 'Karl 2', 'email': 'karl@gmail.com', '_type': 'user', '_index': 'user_table', '_id': 2},
{'name': 'Karl 3', 'email': 'karl@gmail.com', '_type': 'user', '_index': 'user_table', '_id': 3},
]
# chunk_size は件数の単位です
deque(es_helpers.parallel_bulk(es, data, thread_count=4, chunk_size=50), maxlen=0)ここで deque について少し紹介します。本来は双方向リストを作るためのものですが、ここでは少し工夫して、その高い性能を利用してジェネレーターを消費する方法として使っています。その働きは、次のコードと同じです。ジェネレーターを最後まで1回通しています。
for response in es_helpers.parallel_bulk(es, data, thread_count=4, chunk_size=50): pass
参考文献：https://docs.python.org/2/library/itertools.html#recipes