【AWS】OpenSearchServiceでWebアプリを実装した件

はじめに
--
AWS Open Search Serviceを使ったことが無かったので、試しに実施した内容・躓いた内容の備忘録です。
OpenSearchとLambdaを活用して、S3に保存されたCSVデータをインデックス化し、全フィールドを対象にしたキーワード検索が可能なWebアプリを構築しました。

システム概要
--
<目的要件>
・S3に保存されたCSVファイルをOpenSearchにインデックス化。
・全フィールドを対象にキーワード検索を可能にする。
・putputは対象のカラム。これらをWebアプリで表示。

<構成>
以下のサービスを使用し簡易で実行
・AWS OpenSearch Service
・Lambda
・S3
・JavaScriptによるWebフロントエンド(SDK dor Javascriptを使用)

<簡易アーキテクチャ>
![キャプチャ.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3978543/50a9c6f9-e11d-ce00-f642-85209155d375.png)


OpenSearchのセットアップ
--
OpenSearch Serviceでドメインを作成し、専用のエンドポイントを取得。
インデックス名: hogehoge を作成。
<確認方法>
　・OpenSearch Dashboardsが利用可能な場合、そこからインデックスの作成とデータ確認を実施。
　・Dashboardsが利用できない場合、以下のLambda関数を実行して確認

```
def get_indices():
    url = f"{OPENSEARCH_ENDPOINT}/_cat/indices?v"
    headers = {"Content-Type": "application/json"}
    response = requests.get(url, auth=auth, headers=headers)
    if response.status_code == 200:
        return {
            "statusCode": 200,
            "body": json.dumps({"indices": response.text})
        }
    else:
        return {
            "statusCode": response.status_code,
            "body": json.dumps({"error": response.text})
        }
```


コンソールで設定する場合
--
手順 1: OpenSearchのドメイン作成画面に移動
　AWS Management Consoleにログイン
　検索バーで「OpenSearch」と入力し、サービスにアクセス
　左側メニューから「ドメイン」を選択し、「ドメインを作成」ボタンをクリック

手順 2: 基本設定を入力
　以下の情報を入力します。(各自カスタマイズしてください)

　名前：作成するドメインの名前を入力（例: mynewdomain）
　ドメインの作成方法：簡単作成
　エンジンオプション：最新バージョン（例: OpenSearch 2.17）を選択
　データノード：r7g.large.search（3ノード、300GiBストレージ）
　専用マスターノード：m7g.large.search（3ノード）
　ネットワーク：VPCアクセス


手順 3: ネットワーク設定
　ネットワークアクセスを選択：VPCアクセス
　必要に応じて、デュアルスタックモード（IPv4, IPv6対応）を選択
　VPC：使用するVPCを選択


手順 4: アクセスコントロール
　マスターアクセスの設定:
　IAMユーザーのARNを設定する（例: arn:aws:iam::<account-id>:role/my-administrator）。
　必要に応じて、IAMユーザーを新規作成

手順 5: デフォルト設定の確認
　以下の設定を確認し、必要に応じて変更

　デプロイタイプ/可用性: スタンバイ付きのマルチAZ
　データノードインスタンスタイプ：r7g.large.search
　アベイラビリティゾーン：3-AZ
　「作成」をクリックすると、ドメインが作成されます。20分くらいかかりました。
 

インデックス名を設定する
--
以下のコードを用いてAWS Lambdaで定義した。(CLI等でやるのがセオリーだと思う)
今回はS3に置いているcsvファイル内を検索したかったのでそのあたりを設定。

```
def index_s3_data():
    bucket_name = "your_bucket_name"  # S3バケット名
    file_key = "path/to/your.csv"  # S3内のファイルパス

    # S3からCSVデータを取得
    response = s3.get_object(Bucket=bucket_name, Key=file_key)
    csv_content = response['Body'].read().decode('utf-8')
    rows = csv.DictReader(csv_content.splitlines())

    # OpenSearchにデータを登録
    url = f"{OPENSEARCH_ENDPOINT}/{INDEX_NAME}/_doc"
    headers = {"Content-Type": "application/json"}
    for row in rows:
        response = requests.post(url, auth=auth, headers=headers, data=json.dumps(row))
        if response.status_code not in [200, 201]:
            print(f"Failed to index document: {response.text}")
        else:
            print(f"Document indexed successfully: {row}")

    return {
        "statusCode": 200,
        "body": json.dumps({"message": "Data indexed successfully"})
    }

```




Lambda関数の実装
---

下記Lambdaを設定した
※セキュリティ設定やレイヤーの設定は各自実施してください

```
import boto3
import csv
import json
import requests
from requests_aws4auth import AWS4Auth

# AWS認証情報を取得
region = 'ap-northeast-1'
service = 'es'
session = boto3.Session()
credentials = session.get_credentials()
auth = AWS4Auth(
    credentials.access_key,
    credentials.secret_key,
    region,
    service,
    session_token=credentials.token
)

# OpenSearchエンドポイントとインデックス名
OPENSEARCH_ENDPOINT = "https://<your-opensearch-endpoint>"
INDEX_NAME = "my-new-index"

def lambda_handler(event, context):
    try:
        action = event.get("action", "default")

        if action == "index_data":
            return index_s3_data()
        elif action == "search":
            query = event.get("query")
            if not query:
                return {
                    "statusCode": 400,
                    "body": json.dumps({"error": "Query parameter is missing"})
                }
            return search_data(query)
        else:
            return {
                "statusCode": 400,
                "body": json.dumps({"message": "Invalid action specified"})
            }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({"message": "Error occurred", "error": str(e)})
        }
```



フロントエンド (HTML + JavaScript) の実装
--
HTML (index.html)
```
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>OpenSearch App</title>
    <script src="https://sdk.amazonaws.com/js/aws-sdk-2.1232.0.min.js"></script>
</head>
<body>
    <h1>Search S3 Data using OpenSearch</h1>
    <div>
        <label for="searchQuery">Enter Search Keyword:</label>
        <input type="text" id="searchQuery" placeholder="Search...">
        <button id="searchButton">Search</button>
    </div>
    <div id="results">
        <h2>Results:</h2>
        <ul id="resultsList"></ul>
    </div>
    <script src="script.js"></script>
</body>
</html>
```

JavaScript (script.js)
```
AWS.config.update({
    region: "ap-northeast-1",
    credentials: new AWS.Credentials("<ACCESS_KEY>", "<SECRET_KEY>")
});

const lambda = new AWS.Lambda();

const searchButton = document.getElementById("searchButton");
const searchQueryInput = document.getElementById("searchQuery");
const resultsList = document.getElementById("resultsList");

function addMessage(message, isError = false) {
    const messageItem = document.createElement("li");
    messageItem.textContent = message;
    messageItem.style.color = isError ? "red" : "black";
    resultsList.appendChild(messageItem);
}

function callLambda(query, callback) {
    const payload = { action: "search", query: query };
    const params = {
        FunctionName: "opensearch_test",
        InvocationType: "RequestResponse",
        Payload: JSON.stringify(payload)
    };

    lambda.invoke(params, (err, result) => {
        if (err) {
            callback(`Error: ${err.message}`, true);
        } else {
            const data = JSON.parse(result.Payload);
            if (data.statusCode === 200) {
                const body = JSON.parse(data.body);
                callback(body.results || []);
            } else {
                callback(`Lambda Error: ${data.body}`, true);
            }
        }
    });
}

searchButton.addEventListener("click", () => {
    const query = searchQueryInput.value.trim();
    if (!query) {
        alert("Enter a search query.");
        return;
    }
    resultsList.innerHTML = "";
    callLambda(query, (response, isError = false) => {
        if (isError) {
            addMessage(response, true);
        } else {
            response.forEach(result => addMessage(result));
        }
    });
});
```


結果
--
ローカルで検索ができることを確認できた。
(outputはファイル名にしたので問題ないことが確認できた。)
(デザイン面サボってます。)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3978543/403a88b9-77ae-5055-bf80-67f2905af1ca.png)



実装時につまずいたポイント
--

①インデックス確認ができない問題 / 検索結果が空になる問題
<エラー内容>
正しくインデックス化されたと思われるデータに対し、検索しても結果が返らず[]（空配列）になる。

<原因>
・インデックスネームが誤っている場合
・検索クエリで"fields": ["*"]が設定されていない場合

<解決策>
上記get_indices()関数を使用してインデックスネームを取得し、正しいインデックスネームを設定する。

②レスポンスの解析エラー
<エラー内容>
レスポンスデータが予想した形式と異なり、JavaScriptでエラーが発生。
```
レスポンス解析エラー: TypeError: Cannot convert undefined or null to object
```


<発生箇所>
フロントエンドのcallLambda関数で、レスポンスのresult._sourceが正しく処理されない場合

<解決策>
データ構造を確認し、文字列のレスポンスをそのまま表示するよう修正。
```
if (response.length > 0) {
    response.forEach(result => {
        addMessage(result);  // レスポンスが文字列の場合、そのまま表示
    });
} else {
    addMessage("該当する結果が見つかりません。");
}
```


③OpenSearchへのデータ登録失敗
<エラー内容>
OpenSearchへのデータ登録時にエラーが発生し、以下のようなログが表示される。
```
Failed to index document: {"error": "...", "status": 400}
```

<発生箇所>
requests.postでのデータ登録部分
<原因>
・フィールドに不要な空白や特殊文字が含まれている
・必須フィールドが不足している

<解決策>
データをインデックス化する前に、すべてのフィールドをクリーニング。
```
document = {key: value.strip() if value else "" for key, value in row.items()}
```


④フロントエンドでのデータ取得失敗
<エラー内容>
ブラウザで検索結果が以下のように表示される:
```
該当する結果が見つかりません。
Folder: 未定義, Description: 未定義
```

<発生箇所>
JavaScriptでresult._sourceを正しく取得できていない場合

<解決策>
Lambdaから返されるレスポンスデータの形式を修正し、必要なフィールドだけを返すように変更
```
report_names = [hit["_source"].get("報告書", "No report name") for hit in hits]
return {
    "statusCode": 200,
    "body": json.dumps({"results": report_names})
}
```

⑤権限エラー
<エラー内容>
Lambdaの実行時にOpenSearchまたはS3へのアクセスが拒否される。
```
AccessDeniedException: User is not authorized to perform: ...
```

<発生箇所>
OpenSearchまたはS3への接続処理
<解決策>
LambdaのIAMロールに以下の権限を追加。
```
AmazonS3ReadOnlyAccess
AmazonOpenSearchServiceFullAccess
```