Elasticsearch Bulk APIの性能を測ってみた


# はじめに

Elasticsearchには、ドキュメント更新を1クエリで実行できる[Bulk API]なんてものがあります。テストデータを投入するときとかに便利なので、使っている方も多いかと思います。
アプリからElasticsearchにデータを投入するとき、Fluentdとかを経由すると自動で[Bulk API]を使ってくれるのですが、直にデータを投入するときは面倒だから1つずつで良いや、とかなったりします。性能差が大きいようなら積極的に[Bulk API]を使った方が良いよねということで、測ってみました。

# 測定条件
適当に、こんな感じでやってみます。
- インデックスは1つだけ。
- ドキュメントは10フィールド * 50文字。

# テストコード
Pythonと[公式クライアント](http://elasticsearch-py.readthedocs.org/en/master/)の組み合わせで、テストコードを書きます。
Bulkサイズを変えながら、10,000件のドキュメントを登録します。

```py
import time
from elasticsearch.client import Elasticsearch
from elasticsearch.exceptions import TransportError
from elasticsearch.helpers import bulk


def generate_test_doc():
    """
    すごい適当にデータを作る。
    """
    return {
        "field1": "aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd",
        "field2": "aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd",
        "field3": "aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd",
        "field4": "aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd",
        "field5": "aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd",
        "field6": "aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd",
        "field7": "aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd",
        "field8": "aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd",
        "field9": "aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd",
        "field10": "aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd",
    }


class PerformanceTimer(object):

    def __init__(self, message):
        self._message = message

    def __enter__(self):
        self._start = time.clock()

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("| {0:30} | {1:>6.3f} |".format(
                self._message,
                (time.clock() - self._start)))


if __name__ == "__main__":
    INDEX_NAME = "test_index"
    TYPE_NAME = "test_type"
    NUM_DOCS = 10000

    es = Elasticsearch([{"host": "192.168.99.100", "port": 9200}])

    try:
        es.indices.delete(INDEX_NAME)
        es.indices.flush()
    except TransportError:
        pass

    # Index API (create)
    with PerformanceTimer("Index API (create)"):
        for i in range(0, NUM_DOCS):
            es.create(INDEX_NAME, TYPE_NAME, generate_test_doc())
        es.indices.flush()

    assert es.count(INDEX_NAME, TYPE_NAME)["count"] == NUM_DOCS
    es.indices.delete(INDEX_NAME)
    es.indices.flush()

    # Bulk API (create)
    for bulk_size in [1, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 120]:
        with PerformanceTimer("Bulk API {0} (create)".format(bulk_size)):
            for i in range(0, NUM_DOCS - (NUM_DOCS % bulk_size), bulk_size):
                actions = []
                for j in range(0, bulk_size):
                    actions.append({
                        "_op_type": "create",
                        "_index": INDEX_NAME,
                        "_type": TYPE_NAME,
                        '_source': generate_test_doc()
                    })
                bulk(es, actions)

            # あまり。
            if NUM_DOCS % bulk_size != 0:
                actions = []
                for j in range(0, NUM_DOCS % bulk_size):
                    actions.append({
                        "_op_type": "create",
                        "_index": INDEX_NAME,
                        "_type": TYPE_NAME,
                        '_source': generate_test_doc()
                    })
                bulk(es, actions)

            es.indices.flush()

        assert es.count(INDEX_NAME, TYPE_NAME)["count"] == NUM_DOCS
        es.indices.delete(INDEX_NAME)
        es.indices.flush()
```

# 測定結果
結果はこんな感じ。

| op                                       | time  |
|-----------------------------|-------|
| Index API (create)             | 23.365 |
| Bulk API 1 (create)            | 20.791 |
| Bulk API 10 (create)           |  6.414 |
| Bulk API 20 (create)           |  4.148 |
| Bulk API 30 (create)           |  3.403 |
| Bulk API 40 (create)           |  3.017 |
| Bulk API 50 (create)           |  2.638 |
| Bulk API 60 (create)           |  2.474 |
| Bulk API 70 (create)           |  2.424 |
| Bulk API 80 (create)           |  2.258 |
| Bulk API 90 (create)           |  2.244 |
| Bulk API 100 (create)          |  2.071 |
| Bulk API 110 (create)          |  2.073 |
| Bulk API 120 (create)          |  1.998 |

![bulktime.png](https://qiita-image-store.s3.amazonaws.com/0/73867/ceea588a-d99c-75db-cad6-68743580f126.png)

# 結論
Elasticsearchにデータを入れるときは[Bulk API]を使いましょう。
一度に入れるドキュメント数は、100くらいで十分。

[Bulk API]: https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html