Implante a lógica de pré-processamento em um modelo de ML em um único endpoint usando um pipeline de inferência na Amazon SageMaker

Prepare o conjunto de dados para sua tarefa de regressão.

Abra um caderno no Amazon SageMaker Studio.

Para importar todas as bibliotecas necessárias e inicializar seu ambiente de trabalho, use o código de exemplo a seguir em seu notebook:

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"

Para baixar um conjunto de dados de exemplo, adicione o seguinte código ao seu caderno:

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

Observação: o exemplo deste padrão usa o Abalone Data Set do UCI Machine Learning Repository.

Faça upload do conjunto de dados em um bucket do S3.

No caderno em que você preparou seu conjunto de dados anteriormente, adicione o código a seguir para carregar seus dados de amostra em um bucket do S3:

WORK_DIRECTORY = "abalone_data"

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

Prepare o script preprocessor.py.

Crie o objeto do pré-processador do SKLearn.

Para criar um objeto pré-processador do skLearn (chamado skLearn Estimator) que você possa incorporar ao seu pipeline de inferência final, execute o seguinte código em seu notebook: 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})

Teste a inferência do pré-processador.

Para confirmar se seu pré-processador está definido corretamente, inicie um trabalho de transformação em lote inserindo o seguinte código em seu SageMaker notebook:

# 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

Criar um objeto modelo.

Para criar um objeto de modelo com base no algoritmo linear do aluno, insira o seguinte código em seu SageMaker caderno:

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)

O código anterior recupera a imagem do Docker do Amazon ECR do Registro público do Amazon ECR para o modelo, cria um objeto estimador e, em seguida, usa esse objeto para treinar o modelo de regressão.

Implantar o modelo de pipeline.

Para criar um objeto de modelo de pipeline (ou seja, um objeto de pré-processador) e implantar o objeto, insira o seguinte código em seu SageMaker notebook:

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)

Observação: você pode ajustar o tipo de instância usado no objeto de modelo para atender às suas necessidades.

Teste a inferência

Para confirmar se o endpoint está funcionando corretamente, execute o seguinte exemplo de código de inferência em seu SageMaker notebook:

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))