CFnによるOpenSearch環境構築サンプルとPythonによるインデックス作成と検索

簡易な構成でOpenSearchを構築し、Pythonでインデックス作成と検索を行うサンプルです。
環境構築はCloudFormationを利用します。

OpenSearchの料金は高めなので、インスタンスは最小のものを選択し
不要になったら削除しましょう。

CloudFormationを利用したOpenSearchの構築
OpenSearchのインデックス作成と検索(Python)

CloudFormationを利用したOpenSearchの構築

パブリックアクセスを有効にし、IAMとIPアドレスによる制限をかけています。

AWSTemplateFormatVersion: "2010-09-09"
Description: OpenSearch Test

Resources:
  OpenSearchServiceDomain:
    Type: AWS::OpenSearchService::Domain
    Properties:
      DomainName: 'dev-demo'
      EngineVersion: 'OpenSearch_1.0'
      AdvancedSecurityOptions:
        Enabled: false
        InternalUserDatabaseEnabled: false

      AccessPolicies:
        Version: '2012-10-17'
        Statement:
          -
            Effect: 'Deny'
            Principal: '*'
            Action: 'es:*'
            Resource: !Sub arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/dev-demo/*
            Condition:
              NotIpAddress:
                aws:SourceIp:
                - xx.xx.xx.xx
          -
            Effect: 'Allow'
            Principal:
              AWS: 'arn:aws:iam::youraccountcode:user/iamusername'
            Action: 'es:*'
            Resource: !Sub arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/dev-demo/*

OpenSearchのインデックス作成と検索(Python)

ローカルマシンからAWSのクレデンシャルを利用し接続します。
接続には「AWS4Auth」ライブラリを利用しています。

ホスト名に指定する値ですが、ドメインエンドポイントから
「https://」を除いた値を指定してください。

from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
import boto3

host = 'yourhostname'
region = 'ap-northeast-1'
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)

search = OpenSearch(
    hosts = [{'host': host, 'port': 443}],
    http_auth = awsauth,
    use_ssl = True,
    verify_certs = True,
    connection_class = RequestsHttpConnection
)

document = {
    "title": "Moneyball",
    "director": "Bennett Miller",
    "year": "2011"
}

# インデックス作成
search.index(index="movies", doc_type="_doc", id="5", body=document)

# 検索実行
print(search.get(index="movies", doc_type="_doc", id="5"))

リンク

以上