Python Elasticsearch 基本的な使い方まとめ

# はじめに
**PythonでElasticsearch**を使う機会があったため情報を収集していましたが、サイトで使われているElasticsearchの**バージョンが古**かったり、そもそも、**情報が少**なかったりしたので、今回、メモとして**簡単な例**と共に基本的な使い方をまとめました。

この記事を読むことで、低レベルクライアント([elasticsearch-py](https://github.com/elastic/elasticsearch-py))を使った**インデックス**および**ドキュメント**の**CRUD**操作ができるようになります。

# 環境
前提として、Elasticsearchはすでに立ち上がっていることとします。

また、実行環境は以下の通りです。特殊なライブラリを使っていないので、どの環境でも基本的には問題ないと思います。

- **利用環境**

```bash:Ubuntuバージョン
Ubuntu 20.04 LTS
```

- **Elasticsearch環境**

```bash:Elasticserchバージョン
7.7.0
```

- **Python環境**

```bash:Pythonバージョン
$ python -V
Python 3.8.4
```

# 準備
- pipでElasticsearchクライアントのインストール

```bash
$ python -m pip install elasticsearch
```

# 基本操作
ここでは、基本的なPythonからElasticsearchを操作する方法を紹介します。

## 接続
インストールしたelasticsearchパッケージを利用してElasticsearchが起動しているホスト(localhost)へ接続する方法は以下の通りです。

### 基本的な接続方法
特にElasticsearch側で認証を設定していなければ、この方法で接続できます。

```python:サンプル
from elasticsearch import Elasticsearch

# Elasticsearchクライアント作成
es = Elasticsearch("http://localhost:9200")
```

ポートや、httpまたはhttpsを指定することもできます。

```python:サンプル
# Elasticsearchインスタンスを作成
es = Elasticsearch(
    ["localhost", "otherhost"],
    scheme="http",
    port=9200
)
```

### HTTP認証を利用した接続
ElasticsearchにIDやパスワードを設定している場合は、この方法で接続できます。

```python:サンプル
es = Elasticsearch(
    "http://localhost:9200",
    http_auth=("user_id", "password")
)
```

**user_id**や**password**は設定されたID、パスワードを表しています。

### 接続の解除
上記で確立した内部接続を**close()**で閉じることができます。

```python:サンプル
# Elasticsearchインスタンスを作成
es = Elasticsearch("http://localhost:9200")

# 内部接続を閉じる
es.close()
```

余談ですが、**close()**しないと、インスタンスが**ガーベージコレクション**された際に、**例外が発生**します。これを防ぐためにも、明示的に書いたほうが良いでしょう。
> This warning is created by aiohttp when an open HTTP connection is garbage collected. You’ll typically run into this when closing your application. To resolve the issue ensure that close() is called before the AsyncElasticsearch instance is garbage collected.
https://elasticsearch-py.readthedocs.io/en/master/async.html?highlight=close#receiving-unclosed-client-session-connector-warning

## インデックスの基本操作
インデックスを扱うための基本的な操作について紹介します。

インデックスの操作には、**indices**という属性を使います。

### インデックスの作成
studentsというインデックスを作成します。
タイプやドキュメントが入っていない空のインデックスです。

```python:サンプル
es.indices.create(index='students')
```


#### マッピングを使った方法
データタイプやインデックスの構造を指定して作成できます。

```python:サンプル
mapping = {
    "mappings": {
        "properties": {
            "name": {"type": "text"},
            "age": {"type": "long"},
            "email": {"type": "text"}
        }
    }
}

es.indices.create(index="students", body=mapping)
```

### インデックス情報の収集
ここでは、インデックスの情報を取得する方法を紹介します。

#### インデックス一覧の取得
接続しているElasticsearchでは、どのようなインデックスがあるのか確認したい場合は、**cat**属性の**indices(index="*", h="index")**を利用します。

**indices**はインデックス情報を返すメソッドです。
今回は、全インデックスを一覧で取得したいので、引数の**index**には**ワイルドカード**を指定します。

また、引数**h**には、**列名**を指定することで**列情報**が改行区切りで返ってきます。

```python:サンプル
# インデックス一覧の取得
indices = es.cat.indices(index='*', h='index').splitlines()
# インデックスの表示
for index in indices:
    print(index)
```

実行結果は、以下の通りです。

```bash:実行結果
.apm-custom-link
.kibana_task_manager_1
.apm-agent-configuration
students
.kibana_1
```

作成した**students**だけではなく、デフォルトで入っているインデックスも表示されました。

#### インデックスのマッピングの確認
特定のインデックスのマッピングを確認する場合は、**get_mapping(index="インデックス名")**を利用します。

```python:サンプル
print(es.indices.get_mapping(index="students"))
```

実行結果は以下の通りです。

```bash:実行結果
{'students': {'mappings': {'properties': {'age': {'type': 'long'}, 'email': {'type': 'text', 'fields': {'keyword': {'type': 'keyword', 'ignore_above': 256}}}, 'name': {'type': 'text', 'fields': {'keyword': {'type': 'keyword', 'ignore_above': 256}}}}}}}
```

また、引数にindexを指定しない場合、**全インデックスのマッピング**を取得することができます。

```python:サンプル
print(es.indices.get_mapping())
```
実行結果は割愛します。

### インデックスの更新
マッピングを変更してインデックスの構造を更新する場合は、**put_mapping(index="インデックス名", body="変更分のマッピング")**を利用します。

例えば、studentsに新たに学籍番号を追加する場合は以下のようになります。

```python:サンプル
mapping = {
    "properties": {
        "student_number": {"type": "long"}
    }
}

es.indices.put_mapping(index="students", body=mapping)
```
指定するマッピングは全て渡す必要はなくて、**差分**だけで良いです。

また、インデックス作成の際には**mappings**配下にネストしましたが、更新の場合は、そうではないことに注意しましょう。

現在のマッピングは、以下のようになっています。

```bash:現在のマッピング
{
  "students": {
    "mappings": {
      "properties": {
        "age": {
          "type": "long"
        },
        "email": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "name": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "student_number": {
          "type": "long"
        }
      }
    }
  }
}
```

見ずらいですが、最後に新たに追加した**student_number**が表示されています

### インデックスの削除
特定のインデックスを削除したい場合は、**delete(index="インデックス名")**を利用します。

今回は、今まで操作してきた**students**を消してみます。

```python:サンプル
es.indices.delete(index="students")
```

### インデックスの存在確認
エラーハンドリングを追加するにあたって、インデックスが存在するか確認したい。

そのような場合は、**exists(index="インデックス名")**を利用します。

```python:サンプル
print(es.indices.exists(index="students"))
```

実行結果は、以下のようになります

```bash:実行結果
False
```

先程、インデックスを消したので**False**が返ってきました。
存在する場合は、**True**が返ってきます。

## ドキュメントの基本操作
次は、ドキュメントを扱うための基本的な操作について紹介します。

### ドキュメントの作成
ドキュメントを新しく登録するには、**create(index="インデックス名", id="ドキュメントID", body="新規ドキュメント")**を利用します。

※ インデックスが無い場合は、登録するドキュメントから自動的に型を判断して作成されます。

```python:サンプル
# 登録したいドキュメント
student = {
    "name": "Taro",
    "age": 36,
    "email": "taro@example.com"
}

# ドキュメントの登録
es.create(index='students', id=1, body=student)
```

登録に成功すると、以下のような結果が出力されます。

```bash:実行結果
{'took': 1, 'timed_out': False, '_shards': {'total': 1, 'successful': 1, 'skipped': 0, 'failed': 0}, 'hits': {'total': {'value': 0, 'relation': 'eq'}, 'max_score': None, 'hits': []}}
```

#### バルクインサート
**create**を複数回呼び出すことで、複数ドキュメントを登録できますが、**bulk(インスタンス, データ)**を使うことで、一度に登録できます。

引数で渡すデータの構造は、以下の通りで、少々複雑です。

```json
{
  "_op_type": "createやdelete、updateといったアクションを指定"
  "_index": "インデックス名"
  "_id": "ドキュメントのIDを指定(createの場合はなくても良い)"
  "_source": "登録したいドキュメント"
}
```

上記のデータを配列に格納して、**bulk**に渡すことで複数ドキュメントを操作することができます。

今回は、配列ではなく`yield`を使った方法をサンプルとして紹介します。

```python:サンプル
from elasticsearch import Elasticsearch, helpers # bulkを使うために追加


# Elasticsearchインスタンスを作成
es = Elasticsearch("http://localhost:9200")

def gendata():
    # 登録したいドキュメント
    students = [
        {
            "name": "Jiro",
            "age": 25,
            "email": "jiro@example.com"
        },
        {
            "name": "Saburo",
            "age": 20,
            "email": "saburo@example.com"
        }
    ]

    # bulkで扱えるデータ構造に変換します
    for student in students:
        yield {
            "_op_type": "create",
            "_index": "students",
            "_source": student
        }

# 複数ドキュメント登録
helpers.bulk(es, gendata())
```

※ **100MB**を超える**大量のドキュメント**をバルクインサートを実行する場合、エラーが発生します。このようなときは、記事下部に記載した[非同期で大量のドキュメントをバルクインサート](https://qiita.com/satto_sann/private/8a63761bbfd6542bb9a2#%E9%9D%9E%E5%90%8C%E6%9C%9F%E3%81%A7%E5%A4%A7%E9%87%8F%E3%81%AE%E3%83%89%E3%82%AD%E3%83%A5%E3%83%A1%E3%83%B3%E3%83%88%E3%82%92%E3%83%90%E3%83%AB%E3%82%AF%E3%82%A4%E3%83%B3%E3%82%B5%E3%83%BC%E3%83%88)を参照ください。

### ドキュメントの検索
ドキュメントを検索する方法を２つ紹介します。

#### クエリを使って検索
登録したドキュメントを検索するには、**search(index="インデックス名", body="検索クエリ", size=検索数)**を利用します。

**body**や**size**は指定しない場合は、全件表示されます。

```python:サンプル
# ageの値が20より大きいドキュメントを検索するためのクエリ
query = {
    "query": {
        "range": {
            "age": {
                "gt": 20
            }
        }
    }
}

# ドキュメントを検索
result = es.search(index="students", body=query, size=3)
# 検索結果からドキュメントの内容のみ表示
for document in result["hits"]["hits"]:
    print(document["_source"])
```

実行結果は以下のようになります。

```bash:実行結果
{'name': 'Taro', 'age': 36, 'email': 'taro@example.com'}
{'name': 'Jiro', 'age': 25, 'email': 'jiro@example.com'}
```

**Saburo**の歳は20歳なので、検索から外れ表示されていません。

#### IDを使って検索
ドキュメントIDを指定して直接検索する場合は、**get_source(index="インデックス名", id="ドキュメントID")**を利用します。

サンプルでは、idが1のドキュメントを検索しています。

```python:サンプル
print(es.get_source(index="students", id=1))
```

実行結果は以下の通りです。

```bash:実行結果
{'name': 'Taro', 'age': 36, 'email': 'taro@example.com'}
```

idが1の`Taro`が表示されました。

###ドキュメント数の取得
インデックスの中にドキュメント数がいくつあるのか確認したい場合は、**count(index="インデックス名")**を利用します。

```python:サンプル
print(es.count(index="students"))
```

実行結果は以下の通りです。

```bash:実行結果
{'count': 3, '_shards': {'total': 1, 'successful': 1, 'skipped': 0, 'failed': 0}}
```

countの値が**3**となっていいるので、インデックス**students**には現在3つのドキュメントが存在していることがわかります。

### ドキュメントの更新
ドキュメント更新には**update(index="インデックス名", id="ドキュメントID", body="変更内容")**を利用します。


```python:サンプル
# doc配下に変更したいパラメータを記述
student = {
    "doc": {
        "age": 40
    }
}

# ドキュメントIDを指定して更新
es.update(index="students", id=1, body=student)

# 更新されているか確認
print(es.get_source(index="students", id=1))
```

`get_source`を使って更新されているか確認しました。

```bash:実行結果
{'name': 'Taro', 'age': 40, 'email': 'taro@example.com'}
```

ageが40に変更されています。

### ドキュメントの削除
特定のドキュメントを削除する場合は、**delete(index="インデックス名", id="ドキュメントID")**を利用します。

サンプルでは、idが1のドキュメントを削除します。

```python:サンプル
es.delete(index="students", id=1)
```

このドキュメントが本当に削除されているかの確認は次で行います。

### ドキュメントの存在確認
ドキュメントの存在を確認するためには、**exists(index="インデックス名", id="ドキュメントID")**を利用します。

```python:サンプル
print(es.exists(index="students", id=1))
```

```bash:実行結果
False
```

先程、**delete**でidが1のドキュメントを消しているので、**False**が返却され存在しないと表示されました。
存在する場合は、**True**なので、存在と表示されるはずです。

※ 余談ですが、インデックスには存在しているものの、`_source`があるのか調べるには**exists_source**を利用します。

## アドバンス
ここでは、通常あまり使わないようなメソッドや小ネタの紹介をします。

### 非同期で検索
ドキュメント数が何万とある場合、検索に時間がかかってしまいます。
このような場合、`v7.8.0 `からサポートされた**AsyncElasticsearch**を使うことで非同期でリソースを効率的に検索することが出来ます。

#### 準備
機能を使うためには**asyncio**をインストールする必要がありあす。

```bash:インストール
$ python -m pip install elasticsearch [ async ] > = 7 .8.0
```

#### 非同期の検索サンプル
非同期で検索するためのサンプルを用意しました。
**async**や**await**を付け加えただけで、基本的には**search**とやることは変わりません。

```python:サンプル
import asyncio
from elasticsearch import AsyncElasticsearch


# 非同期対応したElasticsearchインスタンスを作成
es = AsyncElasticsearch("http://localhost:9200")


async def main():
    # 非同期検索
    result = await es.search(
        index="students",
        body={"query": {"match_all": {}}},
        size=20
    )
    # 検索結果の表示
    for student in result['hits']['hits']:
        print(student['_source'])

    # セッションをクローズ
    await es.close()

# イベントループを取得
loop = asyncio.get_event_loop()
# 並列に実行して終るまで待つ
loop.run_until_complete(main())
```

動きとしては、

- `asyncio.get_event_loop()`でイベントループを取得
- 並列で動かしたい関数`main()`に`async`をつけて定義
- 時間がかかる処理`search`は`await`をつけて宣言
- イベントループの`run_until_complete` で並列的に実行しつつ終るまで待つ

このような感じです。

### 非同期でバルクインサート
次は、非同期でバルクインサートを実行する方法を紹介します。
コードは**bulk**とほとんど変わらず、**async**や**await**および非同期に対応した**async_bulk**を利用するだけです。

```python:サンプル
import asyncio
from elasticsearch import AsyncElasticsearch
from elasticsearch.helpers import async_bulk

# 非同期対応したElasticsearchインスタンスを作成
es = AsyncElasticsearch("http://localhost:9200")


async def gendata():
    # 登録したいドキュメント
    students = [
        {
            "name": "Siro",
            "age": 19,
            "email": "siro@example.com"
        },
        {
            "name": "Goro",
            "age": 13,
            "email": "goro@example.com"
        }
    ]

    # bulkで扱えるデータ構造に変換します
    for student in students:
        yield {
            "_op_type": "create",
            "_index": "students",
            "_source": student
        }


async def main():
    # 非同期でバルクインサートを実行
    await async_bulk(es, gendata())
    # セッションをクローズ
    await es.close()

# イベントループを取得
loop = asyncio.get_event_loop()
# 並列に実行して終るまで待つ
loop.run_until_complete(main())
```

実行後、登録されたドキュメントを確認すると

```bash:ドキュメント一覧
{'name': 'Jiro', 'age': 25, 'email': 'jiro@example.com'}
{'name': 'Saburo', 'age': 20, 'email': 'saburo@example.com'}
{'name': 'Siro', 'age': 19, 'email': 'siro@example.com'}
{'name': 'Goro', 'age': 13, 'email': 'goro@example.com'}
```

**Saburo**と**Goro**が追加されていることがわかります。

### 非同期で大量のドキュメントをバルクインサート
100MBを超えるドキュメントを一度にバルクインサートすることは出来ません。
これは、Elasticsearch側で上限が設定されているためです。

> **http.max_content_length**
The max content of an HTTP request. Defaults to 100MB.
https://www.elastic.co/guide/en/elasticsearch/reference/current/modules-http.html

このような大量のデータを扱う場合、**async_streaming_bulk(client, actions, その他パラメータ)**を利用して、非同期でドキュメントを複数(チャンク)に分割して登録を行います。

#### パラメータ
指定できるパラメータについて紹介します。

- **chunk_size**<br>
  - 型：整数
  - Elasticsearchへ一度に送信するドキュメント数（デフォルト：500）
- **max_chunk_bytes**<br>
  - 型：整数 
  - リクエストの最大バイトサイズ（デフォルト：100MB）
- **raise_on_error**<br>
  - 型：Bool
  - exceptで`BulkIndexError`発生時に、エラーリストを取得できるようになる（デフォルト: True）
- **raise_on_exception**
  - 型：Bool
  - `False`にすると、bulk失敗時に例外を発生させず、最後に失敗したアイテムの報告のみ行う（デフォルト：True）
- **max_retries**
  - 型：整数
  - ステータスエラー`429`発生時に再試行を行う回数（デフォルト：0）
- **initial_backoff**
  - 型：整数
  - 再試行まで待機する秒数、２回以降は`待機秒数 x 2`となる（デフォルト：2）
- **max_backoff**
  - 型：整数
  - 再試行が待機する最大秒数（デフォルト：600）
- **yield_ok**
  - 型：Bool
  - `False`にすると、出力結果からbulkに成功したドキュメントが表示されなくなる（デフォルト：True） 

#### サンプル
以下に簡単なサンプルを紹介します。

```python:サンプル
import asyncio
from elasticsearch import AsyncElasticsearch
from elasticsearch.helpers import async_streaming_bulk, BulkIndexError

# 非同期対応したElasticsearchインスタンスを作成
es = AsyncElasticsearch("http://localhost:9200")


async def gendata():
    # 登録したいドキュメント
    students = [
        {
            "name": "Siro",
            "age": 19,
            "email": "siro@example.com"
        },
        {
            "name": "Goro",
            "age": 13,
            "email": "goro@example.com"
        }
    ]

    # bulkで扱えるデータ構造に変換します
    for student in students:
        yield {
            "_op_type": "create",
            "_index": "students",
            "_source": student
        }


async def main():
    try:
        # ドキュメントを複数（チャンク）に分けてバルクインサート
        async for ok, result in async_streaming_bulk(client=es,
                                                     actions=gendata(),
                                                     chunk_size=50,  # 一度に扱うドキュメント数
                                                     max_chunk_bytes=52428800  # 一度に扱うバイト数
                                                     ):
            # 各チャンクごとの実行結果を取得
            action, result = result.popitem()
            # バルクインサートに失敗した場合
            if not ok:
                print(f"failed to {result} document {action}")
    # 例外処理
    except BulkIndexError as bulk_error:
        # エラーはリスト形式
        print(bulk_error.errors)

    # セッションのクローズ
    await es.close()

# イベントループを取得
loop = asyncio.get_event_loop()
# 並列に実行して終るまで待つ
loop.run_until_complete(main())
```

今回指定した、チャンクに関するパラメータは、

- チャンク数：50
- チャンクの最大バイト数：50MB

こんな感じです。

これらは、登録したいデータに応じてチューニングすると良いと思います。

### 特定のステータスによるエラーを無視
引数**ignore**に無視したい特定のステータスコードを指定することでエラーを無視することができます。

例えば、存在しないインデックスを削除しようとした場合、**404**エラーが発生します。

```bash:404エラー
elasticsearch.exceptions.NotFoundError: NotFoundError(404, 'index_not_found_exception', 'no such index [test-index]', test-index, index_or_alias)
```

サンプルではこれを無視してみます。

```python:サンプル
# 404と400で発生するエラーを無視
es.indices.delete(index='test-index', ignore=[400, 404])
```

### タイムアウト
es側ではデフォルトでタイムアウトが設定されていますが、自分で設定することも出来ます。

方法は簡単で、引数に**request_timeout=秒数（浮動小数点）**を渡すだけです。
あえて短い秒数を指定して、タイムアウトさせるサンプルを用意しました。

```python:サンプル
print(es.cluster.health(wait_for_status='yellow', request_timeout=0.001))
```

実行すると、指定した0.001秒以内に処理が終了しなかったため、以下のようなタイムアウトのエラーが表示されました。

```bash:タイムアウトエラー
elasticsearch.exceptions.ConnectionError: ConnectionError((<urllib3.connection.HTTPConnection object at 0x7f11297c5520>, 'Connection to localhost timed out. (connect timeout=0.001)')) caused by: ConnectTimeoutError((<urllib3.connection.HTTPConnection object at 0x7f11297c5520>, 'Connection to localhost timed out. (connect timeout=0.001)'))
```


### レスポンスの整形
Elasticsearchからのレスポンスをそのまま表示すると

```bash:レスポンス
{'students': {'mappings': {'properties': {'age': {'type': 'long'}, 'email': {'type': 'text', 'fields': {'keyword': {'type': 'keyword', 'ignore_above': 256}}}, 'name': {'type': 'text', 'fields': {'keyword': {'type': 'keyword', 'ignore_above': 256}}}}}}}
```

このようにとても見づらくなってしまいます。

そこで、**json**パッケージを使って整形することで見やすく表示できます。


```python:サンプル
import json
from elasticsearch import Elasticsearch

# Elasticsearchインスタンスを作成
es = Elasticsearch("http://locahost:9200")

# マッピング情報の取得
response = es.indices.get_mapping(index="students")
# レスポンスの整形
print(json.dumps(response, indent=2))
```

動きとしては、
**json.dumps**に整形したいデータと、インデントの**スペース数**を指定するだけです。

今回はレスポンスのネストが深くなることが予想されたためインデントは**2**を指定しました。

```bash:実行結果
{
  "students": {
    "mappings": {
      "properties": {
        "age": {
          "type": "long"
        },
        "email": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "name": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        }
      }
    }
  }
}
```

見やすく整形されました。

# おわりに
Python Elasticserachのクライアントの基本的な使い方を紹介しました。
直接APIを叩くよりも、用意された便利なメソッドを使うことで、直感的に操作できるのでとても便利に感じました。

使い方が間違っていたり、その他便利な方法があればコメントお願いします！

# 参考サイト
- Python Elasticsearch公式<br>
https://elasticsearch-py.readthedocs.io/en/master/api.html
- PythonでElasticsearchの操作<br>
https://blog.imind.jp/entry/2019/03/08/185935
- 【Python】asyncio（非同期I/O）のイベントループをこねくり回す<br>
https://qiita.com/ynakaDream/items/b63fab24bb30dea6ddb1