Amazon SageMaker 모델 연결

SageMaker에 배포한 표 모델을 Foundry의 외부 모델로 연결합니다. Model Adapter Library에 어댑터를 게시하고 태그를 만든 뒤 연결 정보 및 암호화할 자격 증명을 설정합니다. 저장소가 권장하는 Python 3.10 이상 환경에 boto3, pandas와 palantir_models를 추가하고 선택한 SDK의 지원 조건에 맞게 의존성을 해결합니다.

표 입출력 어댑터

입력 표의 각 행을 features 배열로 바꾸고 instances 안에 넣어 JSON으로 전송합니다. 열 순서는 SageMaker 모델이 기대하는 특성 순서여야 합니다. 응답은 JSON으로 읽어 pandas.json_normalize로 표로 만들 수 있는 구조여야 합니다.

region_name과 endpoint_name은 연결 설정에서, access_key_id와 secret_access_key는 자격 증명에서 읽습니다. Boto3의 SageMaker Runtime 클라이언트를 생성하고 invoke_endpoint에 EndpointName, ContentType 및 Body를 전달합니다.

import palantir_models as pm

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

logger = logging.getLogger(__name__)


class SageMakerTabularAdapter(pm.ExternalModelAdapter):
    """
    :display-name: SageMaker Tabular Model Adapter
    :description: Default model adapter for SageMaker models that expect tabular input and output tabular data.
    """

    def __init__(self, region_name, endpoint_name, access_key_id, secret_access_key):
        self.endpoint_name = endpoint_name
        self.runtime = boto3.client(
            'sagemaker-runtime',
            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.connection_config["region_name"]
        endpoint_name = external_context.connection_config["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 = {
            "instances": df_in.apply(lambda row: {"features": row.tolist()}, axis=1).tolist()
        }
        try:
            response = self.runtime.invoke_endpoint(
                EndpointName=self.endpoint_name,
                ContentType="application/json",
                Body=json.dumps(payload)
            )
        except ClientError as error:
            logger.error("SageMaker inference call failed. This can indicate an error with this Model's egress "
                         "policy. Double check your configured egress policy and ensure the remote endpoint is still "
                         "available.")
            raise error
        try:
            # Output from model is assumed to be json serializable
            # if result is too large for executor this deserialization may cause an OOM
            result = json.loads(response['Body'].read().decode())
        except ValueError as error:
            logger.error("This SageMakerTabularAdapter expects results to be json serializable.")
            raise error
        return pd.json_normalize(result)

SageMaker ClientError는 송신 정책, 원격 엔드포인트의 가용성 및 서비스 오류를 확인하라는 로그를 남긴 뒤 다시 발생시킵니다. 응답이 JSON으로 해석되지 않으면 ValueError를 기록한 뒤 전달합니다. 응답 전체를 메모리로 읽고 표로 바꾸므로 실행기 메모리에 비해 큰 결과는 메모리 부족을 일으킬 수 있습니다.

pm.Pandas()는 열을 고정하지 않은 일반 표 계약입니다. 모델 함수를 게시하려면 필수 열과 자료형을 명시해야 합니다.

송신 정책과 어댑터 선택

예시 리전은 us-east-1입니다. 이 리전의 runtime.sagemaker.us-east-1.amazonaws.com 호스트와 443 포트를 허용하는 정책을 사용합니다. 다른 리전이면 해당 SageMaker Runtime 주소로 맞춥니다.

Connect an externally hosted model에서 게시한 SageMakerTabularAdapter를 선택합니다.

연결 설정

이 어댑터는 URL 필드를 사용하지 않으므로 비워 둡니다. Connection Configuration에는 다음 키를 입력합니다.

Credentials Configuration에는 실제 호출 주체의 access_key_id와 secret_access_key를 저장합니다. 비밀 액세스 키를 암호화하지 않는 연결 설정에 두지 않습니다.

사용

연결한 모델은 Modeling Objective의 온라인 배포 또는 네트워크 송신을 구성한 Python 변환에서 사용할 수 있습니다. 모델 접근 사용자는 저장한 자격 증명을 통해 추론하므로 모델 권한의 위임 범위를 확인합니다. Modeling Objective의 배치 배포와 자동 평가는 지원하지 않습니다.