Amazon S3에서 스트리밍 데이터 로드

디렉터리를 생성합니다. 이 샘플에서는 s3-to-opensearch 이름을 사용합니다.

sample.py라는 디렉터리에서 파일을 생성합니다.

import boto3
import re
import requests
from requests_aws4auth import AWS4Auth

region = '' # e.g. us-west-1
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)

host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com
index = 'lambda-s3-index'
datatype = '_doc'
url = host + '/' + index + '/' + datatype

headers = { "Content-Type": "application/json" }

s3 = boto3.client('s3')

# Regular expressions used to parse some simple log lines
ip_pattern = re.compile('(\d+\.\d+\.\d+\.\d+)')
time_pattern = re.compile('\[(\d+\/\w\w\w\/\d\d\d\d:\d\d:\d\d:\d\d\s-\d\d\d\d)\]')
message_pattern = re.compile('\"(.+)\"')

# Lambda execution starts here
def handler(event, context):
    for record in event['Records']:

        # Get the bucket name and key for the new file
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']

        # Get, read, and split the file into lines
        obj = s3.get_object(Bucket=bucket, Key=key)
        body = obj['Body'].read()
        lines = body.splitlines()

        # Match the regular expressions to each line and index the JSON
        for line in lines:
            line = line.decode("utf-8")
            ip = ip_pattern.search(line).group(1)
            timestamp = time_pattern.search(line).group(1)
            message = message_pattern.search(line).group(1)

            document = { "ip": ip, "timestamp": timestamp, "message": message }
            r = requests.post(url, auth=awsauth, json=document, headers=headers)

region과 host의 변수를 편집합니다.

아직 설치하지 않았다면 pip를 설치한 다음, 새 package 디렉터리에 종속 항목을 설치합니다.

cd s3-to-opensearch

pip install --target ./package requests
pip install --target ./package requests_aws4auth

모든 Lambda 실행 환경에는 Boto3가 설치되어 있으므로 배포 패키지에 이를 포함할 필요가 없습니다.

애플리케이션 코드와 종속 항목을 패키지화합니다.

cd package
zip -r ../lambda.zip .

cd ..
zip -g lambda.zip sample.py

트리거 추가(Add trigger)를 선택하고 S3를 선택합니다.

버킷을 선택합니다.

이벤트 유형(Event type)에서 PUT을 선택합니다.

접두사(Prefix)에는 logs/를 입력합니다.

접미사(Suffix)에는 .log를 입력합니다.

재귀 호출 경고를 확인하고 추가(Add)를 선택합니다.

업로드 원본(Upload from)과 .zip 파일(.zip file)을 선택한 다음, 지시에 따라 배포 패키지를 업로드합니다.

업로드가 완료되면 런타임 설정(Runtime settings)을 변경하고 핸들러(Handler)를 sample.handler로 변경합니다. 이 설정은 트리거 후 실행해야 하는 파일(sample.py)과 메서드(handler)를 Lambda에게 알려 줍니다.