AWS s3 をElasticsearchに取り込んで検索してみる（パート２: NLP Inferenceを追加）

# はじめに
前回の[パート1](https://qiita.com/nobuhikosekiya/items/f58a9c72d3972d3e24cc)ではAWS S3のバケットに入っているファイルをElasticの[S3 Connector Client](https://www.elastic.co/guide/en/enterprise-search/8.7/connectors-s3.html)を使ってElasticsearchにインデックスしてテキスト情報などを取り込みました。

今回は、現在流行りのNLPの話題に乗っかるために、Elasticの[NLP Inference機能](https://www.elastic.co/guide/en/machine-learning/8.7/ml-nlp-inference.html)を試します。
具体的に何ができるようになるかというと、Elasticsearchにインデックスされたデータ（今回はS3バケット内にアップロードされているファイルに含まれるテキスト情報）に対してM/L学習モデルを使ってInference（推論）をかけ、その出力を同じElasticsearchのドキュメントに保存（インデックス）できます。
どのようなM/L学習モデルが使えるかは[こちら](https://www.elastic.co/guide/en/machine-learning/8.7/ml-nlp-model-ref.html)に書かれています。

# MLノードの追加
最初に、ElasticsearchのMLノードが必要です。Elastic Cloudの場合Machine Learning Nodeを数クリックで簡単にデプロイできてしまいます :)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/8f6c97b7-b99e-e7a4-0dc8-033372a8003f.png)
とりあえず低めのスペック（２GB RAM)からスタートしています。あとで簡単に拡張できるので。

# MLモデルのデプロイ
サードパーティ学習モデルをM/Lノードにデプロイするには、ElandというElasticのpython製のツールを使います。以下の情報を見ながら今回の手順を実行していきました。
https://www.elastic.co/guide/en/enterprise-search/8.7/machine-learning-start.html

1: 最初にElandのコードをダウンロードし、コンテナをビルドします。
```
git clone git@github.com:elastic/eland.git
cd eland
docker build -t elastic/eland .
```

2: 次に、以下のパラメータを与えて学習モデルをMLノードにアップロードします。
- 学習モデル（--hub-model-idに指定)
- タスクタイプ（--task-typeに指定。fill_mask, ner, text_classification, text_embedding, zero_shot_classificationから選択）
- アップロード先のElasticsearchのURLとその認証のユーザーパスワード
```
docker run -it --rm --network host \
    elastic/eland \
    eland_import_hub_model \
      --url https://XXX.cloud.es.io:443 \
      -u elastic -p <PASSWORD> \
      --hub-model-id distilbert-base-uncased-finetuned-sst-2-english \
      --task-type text_classification \
      --start
```
 
以下実行例です。
```
~/lab/projects/eland % docker run -it --rm --network host \
    elastic/eland \
    eland_import_hub_model \
      --url https://XXX.es.asia-northeast1.gcp.cloud.es.io:443 \
      -u elastic -p XXX \
      --hub-model-id distilbert-base-uncased-finetuned-sst-2-english \
      --task-type text_classification \
      --start --clear-previous
2023-04-17 11:54:41,453 INFO : Establishing connection to Elasticsearch
2023-04-17 11:54:41,587 INFO : Connected to cluster named 'b54f23853a2c4e6ea34aca4b2381d5a3' (version: 8.7.0)
2023-04-17 11:54:41,588 INFO : Loading HuggingFace transformer tokenizer and model 'distilbert-base-uncased-finetuned-sst-2-english'
Downloading: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 48.0/48.0 [00:00<00:00, 80.6kB/s]
Downloading: 100%|████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 629/629 [00:00<00:00, 415kB/s]
Downloading: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 226k/226k [00:00<00:00, 3.56MB/s]
Downloading: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 255M/255M [00:34<00:00, 7.70MB/s]
/usr/local/lib/python3.9/dist-packages/transformers/models/distilbert/modeling_distilbert.py:214: TracerWarning: torch.tensor results are registered as constants in the trace. You can safely ignore this warning if you use this function to create tensors out of constant variables that would be the same every time you call this function. In any other case, this might cause the trace to be incorrect.
  scores = scores.masked_fill(mask, torch.tensor(-float("inf")))  # (bs, n_heads, q_length, k_length)
2023-04-17 11:55:22,574 INFO : Stopping deployment for model with id 'distilbert-base-uncased-finetuned-sst-2-english'
2023-04-17 11:55:22,593 INFO : Deleting model with id 'distilbert-base-uncased-finetuned-sst-2-english'
/usr/local/lib/python3.9/dist-packages/eland/ml/pytorch/_pytorch_model.py:155: ElasticsearchWarning: The default [remove_binary] value of 'false' is deprecated and will be set to 'true' in a future release. Set [remove_binary] explicitly to 'true' or 'false' to ensure no behavior change.
  self._client.options(ignore_status=404).ml.delete_trained_model(
2023-04-17 11:55:22,902 INFO : Creating model with id 'distilbert-base-uncased-finetuned-sst-2-english'
2023-04-17 11:55:22,930 INFO : Uploading model definition
100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 64/64 [01:26<00:00,  1.36s/ parts]
2023-04-17 11:56:49,674 INFO : Uploading model vocabulary
2023-04-17 11:56:49,804 INFO : Starting model deployment
2023-04-17 11:56:59,317 INFO : Model successfully imported with id 'distilbert-base-uncased-finetuned-sst-2-english'
```
 
3: Trained ModelsメニューでMLモデルがロードされたことを確認します。Stateがstartedになっていない場合、ActionsからStartしてください。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/a8c67c1e-2ace-7dfa-265e-f342c4a7c538.png)


# NLP Inferenceのパイプラインを追加
今回は、S3からElasticsearchにデータを取り込む時のパイプライン処理にNLP Inference処理を追加します。最新のv8.7では、Enterprise Searchメニューから簡単にこれを追加できるようになりました。

前回の[記事](https://qiita.com/nobuhikosekiya/items/f58a9c72d3972d3e24cc)で作ったEnterprise Searchメニューから該当Indexを開きます。PipelinesタブにMachine Learning Inference Pipelinesの項目が見えますが、最初は押せません。Ingest PipelinesをCopy and customizeしないといけないとあります。上にあるCopy and customizeリンクをクリックしましょう。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/edb7082a-2a9a-ee26-13cd-1a88f6752826.png)

そうすると、Add Inference Pipelineが押せるようになりました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/7420e68a-f94e-39ac-b3c3-9b5666cb1d1f.png)

適当なPipeline名をつけ（test001)、先ほどのモデルを選択します。Source fieldとして、Inference対象のFieldとして、S3ファイルのテキストが抽出されたフィールド(=body)を選択します。Target fieldは今回は空のままにします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/34f48d47-cdd7-a40c-9d11-06498dbba049.png)

次の画面ではこのML Inferenceをテストできます。テストするために既にインデックスにあるDocumentのIDが必要なので、別のブラウザタブを開いてEnterprise Search > Content > Indiciesのメニューをたどり、IndexのDocumentを確認し、好きなDocument idをコピーします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/bdf1d4af-8f45-8c69-8fe3-8aa114da33c6.png)

そして、ここのDocument IDにペーストすると、そのDocument情報が左側に読み込まれます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/b8ff7946-7f7f-649a-9e70-6a49690cebd7.png)

Simulate Pipelineをクリックすると、結果が右側に出てきます。Inference(推論）結果としてpredicted_value: POSITIVEという結果が得られました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/d115bf75-cd96-8f8d-0944-5dd399d9a361.png)

最終的なパイプライン定義が確認できるので、Create pipelineで確定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/c35fe4e0-655c-3670-dc9e-c0333de26107.png)

Ingest Pipelineの全体はこのようになっていました。
```
[
  {
    "remove": {
      "field": "ml.inference.test001",
      "ignore_missing": true
    }
  },
  {
    "remove": {
      "field": "test001",
      "ignore_missing": true
    }
  },
  {
    "inference": {
      "field_map": {
        "body": "text_field"
      },
      "model_id": "distilbert-base-uncased-finetuned-sst-2-english",
      "on_failure": [
        {
          "append": {
            "field": "_source._ingest.inference_errors",
            "value": [
              {
                "message": "Processor 'inference' in pipeline 'ml-inference-test001' failed with message '{{ _ingest.on_failure_message }}'",
                "pipeline": "ml-inference-test001",
                "timestamp": "{{{ _ingest.timestamp }}}"
              }
            ]
          }
        }
      ],
      "target_field": "ml.inference.test001"
    }
  },
  {
    "append": {
      "field": "_source._ingest.processors",
      "value": [
        {
          "model_version": "8.7.0",
          "pipeline": "ml-inference-test001",
          "processed_timestamp": "{{{ _ingest.timestamp }}}",
          "types": [
            "pytorch",
            "text_classification"
          ]
        }
      ]
    }
  },
  {
    "set": {
      "copy_from": "ml.inference.test001.predicted_value",
      "description": "Copy the predicted_value to 'test001' if the prediction_probability is greater than 0.5",
      "field": "test001",
      "if": "ctx?.ml?.inference != null && ctx.ml.inference['test001'] != null && ctx.ml.inference['test001'].prediction_probability > 0.5"
    }
  }
]
```

# S3ファイルに対してNLP Inferenceをかける
s3に新たなオブジェクトをアップロードし、Enterprise SearchのIndex画面のSyncボタンで即時同期させます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/f854866e-e41c-c6b5-e04d-c66d1e505c01.png)
そうすると、s3ファイルのテキスト抽出とともに、ML Inference処理が適用され、結果test001: "NEGATIVE"というフィウールどがDocumentに追加されるようになりました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/bca57c7b-5470-4b62-e493-a7c5e68b8a5a.png)


# 2つ目のMLモデル(Named Entity Recognition)を適用
先ほどの学習モデルはtext_classificationだったので、今度はNamed Entity Recognitionタイプのモデルを使ったInferenceをパイプラインに追加してみたいと思います。

新しいモデルをElandでアップします（ここでは実は間違えてtask-typeをtext_classificationのままにしてしまいました。後続手順で修正していますが、そのままのログを残します。）
```
docker run -it --rm --network host \
    elastic/eland \
    eland_import_hub_model \
      --url https://XXX.cloud.es.io:443 \
      -u elastic -p <PASSWORD> \
      --hub-model-id elastic/distilbert-base-uncased-finetuned-conll03-english \
      --task-type text_classification \
      --start
```

しかし、キャパシティ不足で２つのMLモデルのアップロードは失敗してしまいました。
```
elasticsearch.ApiError: ApiError(429, 'status_exception', 'Could not start deployment because no ML nodes with sufficient capacity were found')
```

よって、Elastic CloudでMLノードのメモリを増やします。2GBから4GBに変更。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/493481a7-31df-e4de-e078-bc9cd90eef90.png)

モデルのアップロードはされていたので、Elandは再実行せずこの管理画面からstartを実行して、再トライします。startedになればOKです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/c8601028-ef86-1ca8-01ef-2d7b4d0eb6e3.png)

Memoryの使用状況はこの画面から確認できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/f8ad9222-c459-ac61-a7e9-ebb77de5dd31.png)

test002として、新しくデプロイしたMLモデルのパイプラインを作ります。前と同じくbodyフィールドに対して分析します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/bd2661c1-07f6-2805-8f56-d5a562ed8189.png)

テストしてみたら、predicted valueに"I_MISC" ?? 
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/2e322175-0f93-86a6-9d9a-801ef36d99dc.png)

どうやらMLモデルをアップロードするときにタイプを間違えていたようです。
--task-type text_classification　ではなく
--task-type ner　と設定すべきでした。

先ほどアップロードしたモデルを画面からDeleteします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/0744ca50-bcaf-f8d1-77f1-94cb8727a4b1.png)

再び、Elandでアップロード。nerをtask-typeに指定。
```
docker run -it --rm --network host \
    elastic/eland \
    eland_import_hub_model \
      --url https://XXX.cloud.es.io:443 \
      -u elastic -p <PASSWORD> \
      --hub-model-id elastic/distilbert-base-uncased-finetuned-conll03-english \
      --task-type ner \
      --start
```

今度は正しくなってます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/ded180a2-616a-d093-7735-5cdce89a8e30.png)

今度はテスト結果もそれっぽい値になっています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/0820f234-01e2-b948-912d-123ff43935b5.png)

出来上がったPipeline定義はこうなりました。
```
{
  "description": "",
  "processors": [
    {
      "remove": {
        "field": "ml.inference.test002",
        "ignore_missing": true
      }
    },
    {
      "inference": {
        "field_map": {
          "body": "text_field"
        },
        "model_id": "elastic__distilbert-base-uncased-finetuned-conll03-english",
        "on_failure": [
          {
            "append": {
              "field": "_source._ingest.inference_errors",
              "value": [
                {
                  "message": "Processor 'inference' in pipeline 'test002' failed with message '{{ _ingest.on_failure_message }}'",
                  "pipeline": "test002",
                  "timestamp": "{{{ _ingest.timestamp }}}"
                }
              ]
            }
          }
        ],
        "target_field": "ml.inference.test002"
      }
    },
    {
      "append": {
        "field": "_source._ingest.processors",
        "value": [
          {
            "model_version": "8.7.0",
            "pipeline": "test002",
            "processed_timestamp": "{{{ _ingest.timestamp }}}",
            "types": [
              "pytorch",
              "ner"
            ]
          }
        ]
      }
    }
  ],
  "version": 1
}
```


S3にまた新たなファイルをアップし、Syncでオンデマンドで取り込みを行います。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/f854866e-e41c-c6b5-e04d-c66d1e505c01.png)

結果は... どうなんでしょう。日本の文書だからでしょうか。今回は操作方法の実現目的なので出力結果にはこだわらないでおきます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/c0c9c145-afd4-4cb2-4c47-4417e54785e0.png)

他いくつかの英語の文書で再トライ。

文書１：
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/63c90453-bfe3-0adc-5dd5-2e9db3b9efa1.png)

文書２：
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/5240ad2f-f78c-5700-16ed-90f84fa2f7d6.png)

最後に一気に19個のファイルをS3にアップし、同期処理させてみます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109197/fd05d4eb-d62b-5bf5-5431-1baa44e8e70f.png)

Connector Clientのログを見る限り、57秒でS3からの取り込みとElasticsearchへのインデックス化が完了したようです。
```
[FMWK][13:21:15][INFO] Successfully connected to AWS Server.
[FMWK][13:21:15][INFO] Starting doc lookups
[FMWK][13:22:12][INFO] Job reporting task is stopped.
[FMWK][13:22:12][INFO] [p8Bgj4cBElFcAjC8BShT] Sync done: 19 indexed, 0 deleted. (57 seconds)
```

# 終わり
NLP InferenceをElasticsearchのIngest Pipelineとして簡単に追加することができました。そしてパート１で作成したs3データの取り込み（インジェスト)にその処理を組み入れました。

従来この辺のPipeline設定は自分でコードを書かなくてはなりませんでしたが、v8.7からだいぶ画面の設定ベースで進めることができ、大分使いやすくなったと感じます。

どんなデータに対してどんなモデルを使うか、など今回は適当に選択しましたが、とりあえず一連のフローの実装方法のサンプルができたので、ぜひ皆さんもこれを元に使ってみてほしいと思います。（多分Elasticのs3コネクターを使ったNLP Inferenceのサンプル記事は世界初だと思います）

他の使用例としては、例えばText EmbeddingsモデルのInferenceを使えば、ベクター値をElasticsearchに保存し、あとでElaticsearchのKNN-Searchクエリーにてセマンティック検索ができます。

S3のデータの取り込みとそれに対するMLパイプラインの適用が１日で実装できると考えると、アイディア次第では面白いことができるんじゃないでしょうか :)