Lambda의 트리거를 이용하여 Opensearch에 데이터를 주기적으로 저장하는 방법

3분 분량

콘텐츠 수준: 중급

해당기사에서는 AWS EventBridge와 AWS Lambda를 이용하여 지정된 주기에 AWS Opensearch로 데이터를 보내는 방안에 대해서 설명합니다.

OpenSearch를 사용하면 로그 분석, 애플리케이션 검색, 엔터프라이즈 검색 및 기타 다수의 사용 사례에서 데이터의 수집, 보안, 검색, 집계, 보기 및 분석이 쉬워집니다.또한 OpenSearch가 있으면 원하는 방식으로 사용, 수정, 확장, 수익화, 재판매할 수 있다는 100% 오픈 소스 제품의 이점을 누릴 수 있습니다. [1]

해당 기사에서는 AWS EventBridge를 이용하여 매일 0 시에 AWS Lambda를 실행시켜서 새로운 정보들을 Opensearch에 저장하는 간단한 방법에 대해서 설명해드립니다.

사용조건

AWS Lambda 함수 생성
AWS Lambda 레이어 생성
Amazon EventBridge 생성
Opensearch 도메인 생성

단계1: Lambda 함수 생성

[함수 이름] 원하는 함수이름을 입력합니다.
[런타임],[Python 3.11]로 최신버전의 파이썬으로 지정합니다.
나머지 사항은 그대로 두고 [함수 생성]을 클릭하여 함수를 생성합니다.

여기에 이미지 설명 입력

단계2. 배포 패키지 생성

명령 프롬프트를 열고 my-openseach-function 프로젝트 디렉터리를 만듭니다. [2]

mkdir my-opensearch-function

my-sourcecode-function 프로젝트 디렉터리로 이동합니다.

cd my-opensearch-function

외부 라이브러리를 새package 디렉터리에 설치합니다.

pip3 install --target ./package boto3
pip3 install --target ./package requests
pip3 install --target ./package requests_aws4auth

루트에서 설치된 라이브러리를 포함하는 배포 패키지를 만듭니다. 다음 명령은 프로젝트 디렉터리에my-deployment-package.zip 파일을 생성합니다.

cd package
zip -r ../my-deployment-package.zip .

[Lambda],[추가 리소스].[계층],[계층 생성]을 클릭합니다.
계층 구성에 아래와 같이 이름을 입력하고 이전에 생성한 zip파일을 업로드 합니다.
이전에 만든 Lambda함수를 클릭하고 [Layers], [Add a layer],[AWS layers]을 클릭한뒤 이전에 만든 레이어를 추가합니다.

단계3. Opensearch 보안 설정

도메인 액세스 정책 수정을 아래와 같이 수정합니다. [2]

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "*"
      },
      "Action": "es:*",
      "Resource": "arn:aws:es:us-west-1:123456789012:domain/domain-name/*"
    }
  ]
}

추가적으로 Opensearch 대시보드에 접근하여 [Security], [Roles],[Mapped Access], [all_access]에 Lambda의 역할을 매핑합니다.

단계4. Lambda 트리거 생성

이전에 생성한 Lambda 함수를 클릭합니다.
[트리거 추가],[트리거 구성]에서 아래와 같이 Schedule expression에 cron식을 넣어서 설정합니다.
해당 cron식의 경우 매일 0시 (UTC+0) 에 실행되는 표현식입니다. [3]

여기에 이미지 설명 입력

단계5. Lambad 코드 수정

이전 단계에서 만든 Lambda 함수를 클릭합니다.
Lambda 함수에서 [코드],[코드 소스]에 아래와 같은 코드를 넣고 저장한뒤 [Deploy],[Test]를 수행합니다.
host,region,service, index, datatype은 원하시는 유형으로 넣습니다.

import boto3
import requests
import json

from requests_aws4auth import AWS4Auth

host = <도메인 엔드포인트>
region =<리전>
service = 'es'

def configure(host,region,service):
    host = host
    region = region
    service = service
    credentials = boto3.Session().get_credentials()
    awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
    return awsauth

def send_opensearch(index,datatype,document,awsauth):
    index = index
    datatype = datatype 
    url = host + index + '/' + datatype
    headers = {"Content-Type": "application/json"}
    document = document
    awsauth=awsauth
    r = requests.post(url, auth=awsauth, json=document, headers=headers)

def lambda_handler(event, context):
    awsauth=configure(host,region,service)
    index = 'movies'
    datatype = '_doc'
    url = host + index + '/' + datatype
       
    # API를 사용하는 경우 API를 여기에서 호출합니다. 
    # 호출한 API의 데이터들을 반복문을 적용하여 document에 저장하고 보낼 수 있습니다. 
    
    document = {"title": "sample title", "director": "sample director"}
    send_opensearch(index,datatype,document,awsauth)