S3 버킷에 이미지가 업로드되면 그 이미지를 Lambda가 가져가서 미리 세운 SageMaker 추론 엔드포인트에 던져 결과를 얻는 구조를 만듭니다.

결과를 AWS SNS에 전송하는 방법과 Lambda를 CRON을 사용하여 정기적으로 실행하는 방법도 소개합니다.

이전 준비

S3 버킷 생성

SageMaker 추론 엔드포인트 작성
inference-endpoint 합시다.
만드는 방법은 다음 기사를 참조하십시오.


SNS 주제 만들기

이하를 참고로 만들었습니다 (일본어 대응은 아직 같습니다).

1.Lambda 만들기

トリガーを追加를 클릭합니다.


트리거는 S3를 선택합니다.

아래 이미지와 같이 이벤트 유형은 全てのオブジェクトCreateイベントPrefix는 Pictures/를 (다른 폴더에 업로드하면 무시할 수 있음)
Suffix는 .jpg를 넣습니다 (예 : 텍스트 파일이 올라갈 때 무시할 수 있음)


inference.py

import logging
import boto3
import json
import urllib.parse

# Initialize logger
customer_logger = logging.getLogger(__name__)

client = boto3.client('sagemaker-runtime')

# バケット名
AWS_S3_BUCKET_NAME = 'gazou-ageru-bucket'
s3 = boto3.resource('s3')
bucket = s3.Bucket(AWS_S3_BUCKET_NAME)


def handler(event, context):

    # S3にアップロードされたファイルをオブジェクトとしてGETしたいとき
    # トリガーイベントから渡された情報からS3のバケット名とアップロードされたオブジェクト名を取得する
    bucketname = event['Records'][0]['s3']['bucket']['name']
    bucket = s3.Bucket(bucketname)
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')
    obj = bucket.Object(key)
    objbody = obj.get()['Body'].read()

    response = client.invoke_endpoint(
        EndpointName='inference-endpoint',
        Body= objbody,
        ContentType='image/jpeg',
        Accept='application/json'
    )

    body = response['Body']
    result=json.load(body)
    return(result)



# S3バケット内の全てのオブジェクトに対して実行したい場合
'''
for obj in bucket.objects.all():
    key = obj.key
    body = obj.get()['Body'].read()
'''

Runtime은 Python3.7, 핸들러는 inference.handler로 설정되었습니다.

2. 결과를 AWS SNS로 푸시

이미지처럼 destinationの追加를 누릅니다.

다음과 같이 미리 만든 AWS SNS 주제에 대해 非同期 (Asynchronous) 및 成功時 (On success) 설정으로 저장합니다.

SNS 메시지를 성형하고 싶다면

이 설정이라면 추론 결과 등이 JSON 형식으로 SNS로 푸시됩니다. 메시지를 형성하고 싶다면 람다를 더 쓰지 마십시오. 아래를 참조하십시오.

3. 실행 테스트

다음 방법 중 하나를 사용하여 PC 또는 라즈파이에서 이미지를 업로드합니다.

파이썬 (boto3)으로 할 때의 병아리

S3Upload.py

import boto3

s3= boto3.resource('s3')

def function():
    # 画像ファイルの場合
    data = open(filePath, mode='rb')
    s3.Bucket(bucketName).put_object(Key = s3Key, Body = data)
    # テキストファイルの場合
    data = open(filePath, mode='r')
    s3.Bucket(bucketName).put_object(Key = s3Key, Body = data.read())
    return

# または
def function():
    s3 = boto3.client('s3')
    bucket_name = 'バケット名'
    file_name = 'ローカルのファイルパス'
    key_name = 'S3バケットのキー名'

    s3.upload_file(file_name, bucket_name, key_name)

AWS CLI에서 할 경우

example.sh

aws s3 cp filename.jpg s3://bucket-name

참고:

CloudWatchlogs에 저장되어 있는 Lambda의 log나, SNS에 푸시하고 있는 경우는 SNS에 등록하고 있는 메일 주소나 SMS등을 체크합니다. 메세지 성형의 Lambda를 씹지 않으면 장문의 JSON이 도착할 것입니다.

CRON에 의한 정기 실행

※이것은 이번 예라고 직접 사용할 수 없기 때문에 덤입니다. CRON에서 트리거하면 event로서 S3 버킷의 이름이나 객체 이름을 끌어올 수 없게 되므로 코드를 바꿀 필요가 있습니다.
トリガーを追加를 클릭하고 EventBridge(CloudWatch Events)를 선택합니다.

이미지와 같이 新しいルールの作成를 선택하고 Schedule Expression의 위치에 CRON 형식으로 실행 날짜와 시간을 넣습니다 (이미지라면 毎日午前3時30分에 실행).


참고:

이상입니다.