Executar predições em tempo real com um pipeline de inferência

Você pode usar modelos treinados em um pipeline de inferência para fazer predições em tempo real diretamente sem executar o pré-processamento externo. Ao configurar o pipeline, você pode optar por usar os transformadores de atributos internos já disponíveis no Amazon SageMaker. Ou você pode implementar sua própria lógica de transformação usando apenas algumas linhas de código Spark ou scikit-learn.

O MLeap, um formato de serialização e mecanismo de execução para pipelines de machine learning, oferece apoio a Spark, scikit-learn e TensorFlow para treinar pipelines e exportá-los para um pipeline serializado chamado de MLeap Bundle. Você pode desserializar os pacotes de volta no Spark para o escore no modo em lote ou no runtime do MLeap para fornecer serviços de API em tempo real.

Os contêineres em um pipeline escutam na porta especificada na variável de ambiente SAGEMAKER_BIND_TO_PORT (em vez da 8080). Quando executar um pipeline de inferência, o SageMaker fornece automaticamente essa variável de ambiente aos contêineres. Se essa variável de ambiente não estiver presente, os contêineres padrão usam a porta 8080. Para indicar que o contêiner está em conformidade com esse requisito, use o seguinte comando para adicionar um rótulo ao Dockerfile:

LABEL com.amazonaws.sagemaker.capabilities.accept-bind-to-port=true

Se o seu contêiner precisar escutar em uma segunda porta, escolha uma porta no intervalo especificado pela variável de ambiente SAGEMAKER_SAFE_PORT_RANGE. Especifique o valor como um intervalo inclusivo no formato "XXXX-YYYY", em que XXXX e YYYY são inteiros de vários dígitos. O SageMaker fornece esse valor automaticamente quando você executa o contêiner em um pipeline de vários contêineres.

Criar e implantar um endpoint de pipeline de inferência

O código a seguir cria e implanta um modelo de pipeline de inferência em tempo real com modelos SparkML e XGBoost em série usando o SDK do SageMaker.

from sagemaker.model import Model
from sagemaker.pipeline_model import PipelineModel
from sagemaker.sparkml.model import SparkMLModel

sparkml_data = 's3://{}/{}/{}'.format(s3_model_bucket, s3_model_key_prefix, 'model.tar.gz')
sparkml_model = SparkMLModel(model_data=sparkml_data)
xgb_model = Model(model_data=xgb_model.model_data, image=training_image)

model_name = 'serial-inference-' + timestamp_prefix
endpoint_name = 'serial-inference-ep-' + timestamp_prefix
sm_model = PipelineModel(name=model_name, role=role, models=[sparkml_model, xgb_model])
sm_model.deploy(initial_instance_count=1, instance_type='ml.c4.xlarge', endpoint_name=endpoint_name)

Solicitar inferência em tempo real de um endpoint do pipeline de inferência

O seguinte exemplo mostra como fazer predições em tempo real chamando um endpoint de inferência e transmitindo uma carga útil de solicitações no formato JSON:

import sagemaker
from sagemaker.predictor import json_serializer, json_deserializer, Predictor

payload = {
        "input": [
            {
                "name": "Pclass",
                "type": "float",
                "val": "1.0"
            },
            {
                "name": "Embarked",
                "type": "string",
                "val": "Q"
            },
            {
                "name": "Age",
                "type": "double",
                "val": "48.0"
            },
            {
                "name": "Fare",
                "type": "double",
                "val": "100.67"
            },
            {
                "name": "SibSp",
                "type": "double",
                "val": "1.0"
            },
            {
                "name": "Sex",
                "type": "string",
                "val": "male"
            }
        ],
        "output": {
            "name": "features",
            "type": "double",
            "struct": "vector"
        }
    }

predictor = Predictor(endpoint=endpoint_name, sagemaker_session=sagemaker.Session(), serializer=json_serializer,
                                content_type='text/csv', accept='application/json'

print(predictor.predict(payload))

A resposta que você obtém de predictor.predict(payload) é o resultado da inferência do modelo.

Exemplo de pipeline de inferência do Realtime

É possível executar este exemplo de caderno usando o previsor SKLearn que mostra como implantar um endpoint, executar uma solicitação de inferência e desserializar a resposta. Encontre esse caderno e mais exemplos no repositório de exemplos do Amazon SageMaker no GitHub.