Amazon의 추론 파이프라인을 사용하여 단일 엔드포인트의 ML 모델에 전처리 로직을 배포하십시오. SageMaker

회귀 작업을 위한 데이터 세트를 준비합니다.

Amazon SageMaker 스튜디오에서 노트북을 엽니다.

필요한 모든 라이브러리를 가져오고 작업 환경을 초기화하려면 노트북에서 다음 예제 코드를 사용합니다.

import sagemaker
from sagemaker import get_execution_role

sagemaker_session = sagemaker.Session()

# Get a SageMaker-compatible role used by this Notebook Instance.
role = get_execution_role()

# S3 prefix
bucket = sagemaker_session.default_bucket()
prefix = "Scikit-LinearLearner-pipeline-abalone-example"

샘플 데이터 세트를 다운로드하려면 노트북에 다음 코드를 추가합니다.

! mkdir abalone_data
! aws s3 cp s3://sagemaker-sample-files/datasets/tabular/uci_abalone/abalone.csv ./abalone_data

참고: 이 패턴의 예제에서는 UCI Machine Learning Repository의 Abalone 데이터 세트를 사용합니다.

데이터 세트를 S3 버킷에 업로드합니다.

이전에 데이터 세트를 준비한 노트북에 다음 코드를 추가하여 샘플 데이터를 S3 버킷에 업로드합니다.

WORK_DIRECTORY = "abalone_data"

train_input = sagemaker_session.upload_data(
    path="{}/{}".format(WORK_DIRECTORY, "abalone.csv"),
    bucket=bucket,
    key_prefix="{}/{}".format(prefix, "train"),
)

preprocessor.py 스크립트를 준비합니다.

SKLearn 프리프로세서 객체를 생성합니다.

최종 추론 파이프라인에 통합할 수 있는 SKLearn 프리프로세서 객체 (SKLearn Estimator라고 함) 를 만들려면 노트북에서 다음 코드를 실행하십시오. SageMaker

from sagemaker.sklearn.estimator import SKLearn

FRAMEWORK_VERSION = "0.23-1"
script_path = "sklearn_abalone_featurizer.py"

sklearn_preprocessor = SKLearn(
    entry_point=script_path,
    role=role,
    framework_version=FRAMEWORK_VERSION,
    instance_type="ml.c4.xlarge",
    sagemaker_session=sagemaker_session,
)
sklearn_preprocessor.fit({"train": train_input})

프리프로세서의 추론을 테스트합니다.

프리프로세서가 올바르게 정의되었는지 확인하려면 노트북에 다음 코드를 입력하여 일괄 변환 작업을 시작하십시오. SageMaker

# Define a SKLearn Transformer from the trained SKLearn Estimator
transformer = sklearn_preprocessor.transformer(
    instance_count=1, instance_type="ml.m5.xlarge", assemble_with="Line", accept="text/csv"
)


# Preprocess training input
transformer.transform(train_input, content_type="text/csv")
print("Waiting for transform job: " + transformer.latest_transform_job.job_name)
transformer.wait()
preprocessed_train = transformer.output_path

모델 객체를 생성합니다.

선형 학습기 알고리즘을 기반으로 모델 객체를 생성하려면 노트북에 다음 코드를 입력하십시오. SageMaker

import boto3
from sagemaker.image_uris import retrieve

ll_image = retrieve("linear-learner", boto3.Session().region_name)
s3_ll_output_key_prefix = "ll_training_output"
s3_ll_output_location = "s3://{}/{}/{}/{}".format(
    bucket, prefix, s3_ll_output_key_prefix, "ll_model"
)

ll_estimator = sagemaker.estimator.Estimator(
    ll_image,
    role,
    instance_count=1,
    instance_type="ml.m4.2xlarge",
    volume_size=20,
    max_run=3600,
    input_mode="File",
    output_path=s3_ll_output_location,
    sagemaker_session=sagemaker_session,
)

ll_estimator.set_hyperparameters(feature_dim=10, predictor_type="regressor", mini_batch_size=32)

ll_train_data = sagemaker.inputs.TrainingInput(
    preprocessed_train,
    distribution="FullyReplicated",
    content_type="text/csv",
    s3_data_type="S3Prefix",
)

data_channels = {"train": ll_train_data}
ll_estimator.fit(inputs=data_channels, logs=True)

위의 코드는 모델의 퍼블릭 Amazon ECR 레지스트리에서 관련 Amazon ECR 도커 이미지를 검색하고 예측기 객체를 생성한 다음 해당 객체를 사용하여 회귀 모델을 학습시킵니다.

파이프라인 모델을 배포합니다.

파이프라인 모델 객체 (즉, 전처리기 객체) 를 생성하고 객체를 배포하려면 노트북에 다음 코드를 입력합니다. SageMaker

from sagemaker.model import Model
from sagemaker.pipeline import PipelineModel
import boto3
from time import gmtime, strftime

timestamp_prefix = strftime("%Y-%m-%d-%H-%M-%S", gmtime())

scikit_learn_inferencee_model = sklearn_preprocessor.create_model()
linear_learner_model = ll_estimator.create_model()

model_name = "inference-pipeline-" + timestamp_prefix
endpoint_name = "inference-pipeline-ep-" + timestamp_prefix
sm_model = PipelineModel(
    name=model_name, role=role, models= [scikit_learn_inferencee_model, linear_learner_model]
)

sm_model.deploy(initial_instance_count=1, instance_type="ml.c4.xlarge", endpoint_name=endpoint_name)

참고: 모델 객체에 사용되는 인스턴스 유형을 필요에 맞게 조정할 수 있습니다.

추론을 테스트합니다.

엔드포인트가 제대로 작동하는지 확인하려면 노트북에서 다음 샘플 추론 코드를 실행하세요. SageMaker

from sagemaker.predictor import Predictor
from sagemaker.serializers import CSVSerializer

payload = "M, 0.44, 0.365, 0.125, 0.516, 0.2155, 0.114, 0.155"
actual_rings = 10
predictor = Predictor(
    endpoint_name=endpoint_name, sagemaker_session=sagemaker_session, serializer=CSVSerializer()
)

print(predictor.predict(payload))