Amazon CloudSearchからAmazon Elasticsearch Serviceへ変えました

2019/04/30

全文検索をする必要がありまして、本当はCloudSearchを使い続けたいのですが、今は小規模なので低いインスタンスを使いたくて、Amazon Elasticsearch Serviceに変えました。
データ移行はしていません。
データはソース元から入れ直しました。
変えた理由とAWS Lambda Python3.7でそれぞれに対してデータの追加、更新、検索の違いを記録しておこうと思います。

何に使っているか

Alexaへのリクエストに対して、まずキーワードをマスターデータから全文検索しています。
その全文検索のために使用しています。

Amazon CloudSearchからAmazon Elasticsearch Serviceへ変えた理由

コストです。
アクセス数もまだまだ低いので最低限のコストにすることが目的です。

Amazon CloudSearchのコスト

2019年3月の課金結果です。
バージニア北部で一番低いsearch.m1.smallで$0.059 x 744時間で$43.9です。

Amazon Elasticsearch Serviceのコスト

2019年4月の途中からElasticsearch Serviceに変更したので途中経過です。
上記のCloudSearchと同じ期間条件にしてみると、
バージニア北部t2.smallで $0.036 x 744時間で$26.78ぐらいです。
それとEBSは10GBにしているのでその料金が、$0.135 x 10GBで$1.35です。
両方あわせて、$28.13が見込まれます。

可用性を考えるとノードを複数作って専用マスターノードも作ったりするべきで、それを考えるとCloudSearchの方がコストは低くなりそうなのですが、今回は可用性はおいといてとにかくコスト重視にしています。

Amazon Elasticsearch Service

とにかく作ってみました。
ちなみにCloudSearchはこちらの「Amazon CloudSearchにAWS Lambda(Python)からデータをアップロードする」に書いています。

[新しいドメインの作成]から作成しました。
リザーブドインスタンスもあるのですね。将来的には検討したいです。
今回はオンデマンドにします。

デプロイタイプはカスタムを選択しました。

ドメイン名を設定して、とにかくコスト重視で、1つのAZで1つのインスタンス、インスタンスタイプは最も低いt2.small.elasticsearch、専用マスターインスタンスはなしです。
ストレージはEBSです。
(t2.small.elasticsearchの場合はEBSのみでインスタンスストレージは選択できません。)
サイズは設定できる最小の10GBにしました。

特に隠す必要のないデータですので、ネットワークはパブリックにしました。

ドメインポリシーは次のポリシーにしました。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::123456789012:role/es_role"
      },
      "Action": "es:*",
      "Resource": "arn:aws:es:us-east-1:123456789012:domain/course-search/*"
    }
  ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Effect": "Allow",

"Principal": {

"AWS": "arn:aws:iam::123456789012:role/es_role"

"Action": "es:*",

"Resource": "arn:aws:es:us-east-1:123456789012:domain/course-search/*"

}

]

}

特定のIAMロールからのアクセスだけを許可しています。
このIAMロールはLambdaで使用するIAMロールです。
あらかじめ作っておきます。

これでAmazon Elasticsearch Service側の設定は完了です。

AWS LambdaでAmazon Elasticsearch Serviceを使う

参考にさせていただいたサイト
ありがとうございます！！

モジュール(requests-aws4auth, Elasticsearch Client)の準備

私はMacで作業しました。
pythonというディレクトリを作成して、そこにpipでインストールでしました。

$ pip install requests-aws4auth -t .
$ pip install elasticsearch -t .

$ pip install requests-aws4auth -t .

$ pip install elasticsearch -t .

pythonディレクトリごとzipにしました。

zipファイルをアップロードしてAWS Layersを作成しました。

AWS Lambdaの設定

作成したLayersを設定しておきます。
IAMロールは、Elasticsearch Serviceドメインにアクセス許可したIAMロールを設定しています。

アクセス権限はCloudWatch Logsへのログの書き込みと、このLambdaがDynamoDB StreamsトリガーなのでStreamを読める権限を与えています。
ElasticSearch Serviceへのデータの読み書きはAWSのAPIではなく、ElasticSearchクライアントからのアクセスになるので、IAMロールの権限には含めません。

環境変数

ENDPOINT
Elasticsearch Serviceのエンドポイントを指定しました。
REGION
Elasticsearch Serviceのリージョン。
TZ
Asia/Tokyoを指定しました。

データを登録するAWS Lambdaのコード

import traceback, os, logging.config
from elasticsearch import Elasticsearch, RequestsHttpConnection, helpers
from requests_aws4auth import AWS4Auth

logger = logging.getLogger()
logger.setLevel(logging.INFO)

ENDPOINT = os.environ.get('ENDPOINT', '')
REGION = os.environ.get('REGION', '')


def lambda_handler(event, context):
    try:
        awsauth = AWS4Auth(
            os.environ.get('AWS_ACCESS_KEY_ID', ''),
            os.environ.get('AWS_SECRET_ACCESS_KEY', ''),
            REGION,
            'es',
            session_token=os.environ['AWS_SESSION_TOKEN']
        )

        es = Elasticsearch(
            hosts=[{
                'host': ENDPOINT,
                'port': 443
            }],
            http_auth=awsauth,
            use_ssl=True,
            verify_certs=True,
            connection_class=RequestsHttpConnection,
            timeout=1500
        )

        actions = []

        for record in event['Records']:
            if record['eventName'] == 'REMOVE':
                continue

            course_code = record['dynamodb']['NewImage']['course_code']['S']
            course_name = record['dynamodb']['NewImage']['course_name']['S']

            document = {'name': course_name}

            actions.append({
                '_index': 'course_index',
                '_type': 'course',
                '_id': course_code,
                '_source': document
            })

        if len(actions) == 0:
            return

        response = helpers.bulk(es, actions)
        logger.info(response)

    except:
        logger.error(traceback.format_exc())
        raise Exception(traceback.format_exc())

import traceback, os, logging.config

from elasticsearch import Elasticsearch, RequestsHttpConnection, helpers

from requests_aws4auth import AWS4Auth

logger = logging.getLogger()

logger.setLevel(logging.INFO)

ENDPOINT = os.environ.get('ENDPOINT', '')

REGION = os.environ.get('REGION', '')

def lambda_handler(event, context):

try:

awsauth = AWS4Auth(

os.environ.get('AWS_ACCESS_KEY_ID', ''),

os.environ.get('AWS_SECRET_ACCESS_KEY', ''),

REGION,

'es',

session_token=os.environ['AWS_SESSION_TOKEN']

)

es = Elasticsearch(

hosts=[{

'host': ENDPOINT,

'port': 443

}],

http_auth=awsauth,

use_ssl=True,

verify_certs=True,

connection_class=RequestsHttpConnection,

timeout=1500

)

actions = []

for record in event['Records']:

if record['eventName'] == 'REMOVE':

continue

course_code = record['dynamodb']['NewImage']['course_code']['S']

course_name = record['dynamodb']['NewImage']['course_name']['S']

document = {'name': course_name}

actions.append({

'_index': 'course_index',

'_type': 'course',

'_id': course_code,

'_source': document

})

if len(actions) == 0:

return

response = helpers.bulk(es, actions)

logger.info(response)

except:

logger.error(traceback.format_exc())

raise Exception(traceback.format_exc())

IAMロールを設定することにより、環境変数にAWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKENを取得できます。
これをAWS4Authに設定しまして、Elasticsearchクライアントを作成しました。

複数のインデックス作成なので、actionsの配列にインデックスを作成するアクションを追加していきます。
helpers.bulkで配列化したアクションを実行して複数のインデックスを作成しました。

‘_id’に一意となるキーを指定して、’name’が更新されたときにインデックスが更新できるようにしました。

全文検索するAWS Lambdaのコード

import traceback, logging.config, os
from elasticsearch import Elasticsearch, RequestsHttpConnection, helpers
from requests_aws4auth import AWS4Auth

logger = logging.getLogger()
logger.setLevel(logging.INFO)

ENDPOINT = os.environ.get('ENDPOINT', '')
REGION = os.environ.get('REGION', '')


def lambda_handler(event, context):
    try:
        logger.info(event)

        awsauth = AWS4Auth(
            os.environ.get('AWS_ACCESS_KEY_ID', ''),
            os.environ.get('AWS_SECRET_ACCESS_KEY', ''),
            REGION,
            'es',
            session_token=os.environ['AWS_SESSION_TOKEN']
        )

        es = Elasticsearch(
            hosts=[{
                'host': ENDPOINT,
                'port': 443
            }],
            http_auth=awsauth,
            use_ssl=True,
            verify_certs=True,
            connection_class=RequestsHttpConnection,
            timeout=1500
        )

        response = es.search(
            index="course_index",
            body={
                "query": {
                    "match": {
                        'name': event['query']
                    }
                }
            }
        )

        logger.info(response)
        return response

    except:
        logger.error(traceback.format_exc())
        raise Exception(traceback.format_exc())

import traceback, logging.config, os

from elasticsearch import Elasticsearch, RequestsHttpConnection, helpers

from requests_aws4auth import AWS4Auth

logger = logging.getLogger()

logger.setLevel(logging.INFO)

ENDPOINT = os.environ.get('ENDPOINT', '')

REGION = os.environ.get('REGION', '')

def lambda_handler(event, context):

try:

logger.info(event)

awsauth = AWS4Auth(

os.environ.get('AWS_ACCESS_KEY_ID', ''),

os.environ.get('AWS_SECRET_ACCESS_KEY', ''),

REGION,

'es',

session_token=os.environ['AWS_SESSION_TOKEN']

)

es = Elasticsearch(

hosts=[{

'host': ENDPOINT,

'port': 443

}],

http_auth=awsauth,

use_ssl=True,

verify_certs=True,

connection_class=RequestsHttpConnection,

timeout=1500

)

response = es.search(

index="course_index",

body={

"query": {

"match": {

'name': event['query']

}

)

logger.info(response)

return response

except:

logger.error(traceback.format_exc())

raise Exception(traceback.format_exc())

このAWS Lambdaは今はAPI Gatewayトリガーにしていてチャットボットからのリクエストに対応しています。
リクエストは’query’パラメータで受けています。
‘query’パラメータにはキーワードが与えられるのでそれで検索をしています。

例 : {‘query’: ‘aws’}

ここで作ったAPIに対してのチャットボットからのリクエストが例えば次のようなコードです。

def course_search_master(search_text):
    payload = {
        "query": search_text
    }
    response = requests.get(
        api_endpoint,
        data=json.dumps(payload)
    )
    data = response.json()

    return_text = ''
    if data['hits']['total'] > 0:
        hits_records = data['hits']['hits']
        for record in hits_records:
            return_text += record['_id'] + ':'
            return_text += record['_source']['name'] + '\n\n'

    return return_text

def course_search_master(search_text):

payload = {

"query": search_text

}

response = requests.get(

api_endpoint,

data=json.dumps(payload)

)

data = response.json()

return_text = ''

if data['hits']['total'] > 0:

hits_records = data['hits']['hits']

for record in hits_records:

return_text += record['_id'] + ':'

return_text += record['_source']['name'] + '\n\n'

return return_text

Elasticsearch Serviceのレスポンスの[hits][total]に結果が何件あったかがありますので、結果があったかを判定して、[‘hits’][‘hits’][‘_source’]から値を取得できました。

[参考]CloudSearchで全文検索するAWS Lambdaのコード

CloudSearchでデータのアップロードをするコードは「Amazon CloudSearchにAWS Lambda(Python)からデータをアップロードする」に書いていますのでそちらをご参照ください。

CloudSearchで全文検索するコードは次のようなコードを使っていました。

import traceback, logging.config, json, os,boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

ENDPOINT = os.environ.get('ENDPOINT', '')


def lambda_handler(event, context):
    try:
        logger.info(event)
        domain = boto3.client(
            'cloudsearchdomain',
            endpoint_url=ENDPOINT,
            region_name='us-east-1'
        )

        response = domain.search(
            query=event['query']
        )

        logger.info(response)
        return response

    except:
        logger.error(traceback.format_exc())
        raise Exception(traceback.format_exc())

import traceback, logging.config, json, os,boto3

logger = logging.getLogger()

logger.setLevel(logging.INFO)

ENDPOINT = os.environ.get('ENDPOINT', '')

def lambda_handler(event, context):

try:

logger.info(event)

domain = boto3.client(

'cloudsearchdomain',

endpoint_url=ENDPOINT,

region_name='us-east-1'

)

response = domain.search(

query=event['query']

)

logger.info(response)

return response

except:

logger.error(traceback.format_exc())

raise Exception(traceback.format_exc())

チャットボットからのリクエストは次のようなコードです。
ElasticSearchと若干レスポンスが異なっています。

def course_search_master(search_text):
    payload = {
        "query": search_text
    }
    response = requests.get(
        api_endpoint,
        data=json.dumps(payload)
    )
    data = response.json()

    return_text = ''
    if data['hits']['found'] > 0:
        hits_records = data['hits']['hit']
        for record in hits_records:
            return_text += record['fields']['course_code'][0] + ':'
            return_text += record['fields']['course_name'][0] + '\n\n'

    return return_text