예시: Amazon SageMaker 모델 통합

아래 문서는 Amazon SageMaker에 호스팅된 모델에 대한 사용자 지정 연결에 대한 예시 설정 및 모델 어댑터를 제공합니다.

단계별 가이드는 모델 어댑터 생성 방법 및 외부에 호스팅된 모델에 대한 연결 생성 방법에 대한 문서를 참조하십시오.

Amazon SageMaker 테이블형 모델 어댑터 예시

먼저, Code Repositories 애플리케이션에서 모델 어댑터 라이브러리를 사용하여 모델 어댑터를 공개하고 태그합니다. 아래 모델 어댑터는 AWS SDK for Python (Boto3) 및 프레임워크를 사용하여 Amazon SageMaker에 호스팅된 모델에 대한 연결을 설정합니다. 아래 코드는 Python 3.8.17, boto3 1.28.1 그리고 pandas 1.5.3 버전에서 테스트되었습니다.

이 모델 어댑터는 다음 가정을 기반으로 합니다:

이 모델 어댑터는 이 모델에 제공되는 데이터가 테이블형임을 가정합니다.
이 모델 어댑터는 입력 데이터를 JSON으로 직렬화하고 이 데이터를 호스팅된 Amazon SageMaker 모델에 전송합니다.
이 모델 어댑터는 응답이 JSON에서 pandas 데이터프레임으로 역직렬화 가능하다고 가정합니다.
이 모델 어댑터는 Boto3 클라이언트를 구성하기 위해 네 가지 입력을 받습니다.
- region_name - 연결 설정으로 제공됩니다
- endpoint_name - 연결 설정으로 제공됩니다
- access_key_id - 자격 증명으로 제공됩니다
- secret_access_key - 자격 증명으로 제공됩니다

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
import palantir_models as pm
import models_api.models_api_executable as executable_api

import boto3
import json
import pandas as pd
import logging
from typing import Optional
from botocore.exceptions import ClientError

logger = logging.getLogger(__name__)


class SageMakerTabularAdapter(pm.ExternalModelAdapter):
    """
    :display-name: SageMaker 테이블 모델 어댑터
    :description: 탭형 입력을 기대하고 탭형 데이터를 출력하는 SageMaker 모델을 위한 기본 모델 어댑터.
    """

    def __init__(self, region_name, endpoint_name, access_key_id, secret_access_key):
        self.endpoint_name = endpoint_name
        # SageMaker 런타임 클라이언트를 생성합니다.
        self.runtime = boto3.client(
            'runtime.sagemaker',
            aws_access_key_id=access_key_id,
            aws_secret_access_key=secret_access_key,
            region_name=region_name
        )

    @classmethod
    def init_external(cls, external_context) -> "pm.ExternalModelAdapter":
        # 외부 컨텍스트에서 필요한 정보들을 추출합니다.
        region_name = external_context.external_model.connection_configuration["region_name"]
        endpoint_name = external_context.external_model.connection_configuration["endpoint_name"]
        access_key_id = external_context.resolved_credentials["access_key_id"]
        secret_access_key = external_context.resolved_credentials["secret_access_key"]
        return cls(
            region_name,
            endpoint_name,
            access_key_id,
            secret_access_key
        )

    @classmethod
    def api(cls):
        # 입력 및 출력을 정의합니다.
        inputs = {"df_in": pm.Pandas()}
        outputs = {"df_out": pm.Pandas()}
        return inputs, outputs

    def predict(self, df_in):
        # 입력 데이터를 payload로 변환합니다.
        payload = {
            "instances": df_in.apply(lambda row: {"features": row.tolist()}, axis=1).tolist()
        }
        try:
            # SageMaker 엔드포인트를 호출하여 예측을 수행합니다.
            response = self.runtime.invoke_endpoint(
                EndpointName=self.endpoint_name,
                ContentType="application/json",
                Body=json.dumps(payload)
            )
        except ClientError as error:
            logger.error("SageMaker 추론 호출이 실패했습니다. 이는 이 모델의 출구 정책에 오류가 있음을 나타낼 수 있습니다."
                         "구성된 출구 정책을 다시 확인하고 원격 엔드포인트가 여전히 사용 가능한지 확인하세요.")
            raise error
        try:
            # 모델의 출력이 json으로 직렬화 가능한지 가정합니다.
            # 결과가 실행기에 너무 크면 이 역직렬화가 OOM을 일으킬 수 있습니다.
            result = json.loads(response['Body'].read().decode())
        except ValueError as error:
            logger.error("이 SageMakerTabularAdapter는 결과가 json으로 직렬화 가능하다고 예상합니다.")
            raise error
        # 결과를 판다스 데이터프레임으로 변환합니다.
        return pd.json_normalize(result)