Reduza o custo e o tempo de desenvolvimento com o modo local do Amazon SageMaker Pipelines

Criar pipelines de machine learning (ML) robustos e reutilizáveis ​​pode ser um processo complexo e demorado. Os desenvolvedores geralmente testam seus scripts de processamento e treinamento localmente, mas os próprios pipelines geralmente são testados na nuvem. Criar e executar um pipeline completo durante a experimentação adiciona sobrecarga e custos indesejados ao ciclo de vida do desenvolvimento. Neste post, detalhamos como você pode usar Modo local do Amazon SageMaker Pipelines para executar pipelines de ML localmente para reduzir o desenvolvimento de pipeline e o tempo de execução, ao mesmo tempo em que reduz os custos. Depois que o pipeline tiver sido totalmente testado localmente, você poderá executá-lo facilmente com Amazon Sage Maker recursos gerenciados com apenas algumas linhas de alterações de código.

Visão geral do ciclo de vida do ML

Um dos principais impulsionadores de inovações e aplicativos em ML é a disponibilidade e a quantidade de dados, juntamente com opções de computação mais baratas. Em vários domínios, o ML provou ser capaz de resolver problemas anteriormente insolúveis com big data e técnicas analíticas clássicas, e a demanda por profissionais de ciência de dados e ML está aumentando constantemente. De um nível muito alto, o ciclo de vida de ML consiste em muitas partes diferentes, mas a construção de um modelo de ML geralmente consiste nas seguintes etapas gerais:

1. Limpeza e preparação de dados (engenharia de recursos)
2. Treinamento e ajuste de modelos
3. Avaliação de modelo
4. Implantação do modelo (ou transformação em lote)

Na etapa de preparação de dados, os dados são carregados, massageados e transformados no tipo de entradas ou recursos que o modelo de ML espera. Escrever os scripts para transformar os dados geralmente é um processo iterativo, em que loops de feedback rápidos são importantes para acelerar o desenvolvimento. Normalmente não é necessário usar o conjunto de dados completo ao testar scripts de engenharia de recursos, e é por isso que você pode usar o recurso de modo local de Processamento SageMaker. Isso permite que você execute localmente e atualize o código de forma iterativa, usando um conjunto de dados menor. Quando o código final está pronto, ele é enviado para o trabalho de processamento remoto, que usa o conjunto de dados completo e é executado em instâncias gerenciadas do SageMaker.

O processo de desenvolvimento é semelhante à etapa de preparação de dados para as etapas de treinamento e avaliação do modelo. Os cientistas de dados usam o recurso de modo local do SageMaker Training para iterar rapidamente com conjuntos de dados menores localmente, antes de usar todos os dados em um cluster gerenciado pelo SageMaker de instâncias otimizadas para ML. Isso acelera o processo de desenvolvimento e elimina o custo de execução de instâncias de ML gerenciadas pelo SageMaker durante a experimentação.

À medida que a maturidade de ML de uma organização aumenta, você pode usar Pipelines Amazon SageMaker para criar pipelines de ML que unam essas etapas, criando fluxos de trabalho de ML mais complexos que processam, treinam e avaliam modelos de ML. O SageMaker Pipelines é um serviço totalmente gerenciado para automatizar as diferentes etapas do fluxo de trabalho de ML, incluindo carregamento de dados, transformação de dados, treinamento e ajuste de modelos e implantação de modelos. Até recentemente, você podia desenvolver e testar seus scripts localmente, mas precisava testar seus pipelines de ML na nuvem. Isso tornou a iteração no fluxo e na forma dos pipelines de ML um processo lento e caro. Agora, com o recurso de modo local adicionado do SageMaker Pipelines, você pode iterar e testar seus pipelines de ML da mesma forma que você testa e itera em seus scripts de processamento e treinamento. Você pode executar e testar seus pipelines em sua máquina local, usando um pequeno subconjunto de dados para validar a sintaxe e as funcionalidades do pipeline.

Pipelines SageMaker

O SageMaker Pipelines oferece uma maneira totalmente automatizada de executar fluxos de trabalho de ML simples ou complexos. Com o SageMaker Pipelines, você pode criar fluxos de trabalho de ML com um SDK Python fácil de usar e, em seguida, visualizar e gerenciar seu fluxo de trabalho usando Estúdio Amazon SageMaker. Suas equipes de ciência de dados podem ser mais eficientes e dimensionar mais rapidamente armazenando e reutilizando as etapas do fluxo de trabalho que você cria no SageMaker Pipelines. Você também pode usar modelos pré-criados que automatizam a infraestrutura e a criação do repositório para criar, testar, registrar e implantar modelos em seu ambiente de ML. Esses modelos estão automaticamente disponíveis para sua organização e são provisionados usando Catálogo de serviços da AWS produtos.

O SageMaker Pipelines traz práticas de integração contínua e implantação contínua (CI/CD) para ML, como manter a paridade entre ambientes de desenvolvimento e produção, controle de versão, testes sob demanda e automação de ponta a ponta, o que ajuda você a dimensionar o ML em todo o seu organização. Os praticantes de DevOps sabem que alguns dos principais benefícios do uso de técnicas de CI/CD incluem um aumento na produtividade por meio de componentes reutilizáveis ​​e um aumento na qualidade por meio de testes automatizados, o que leva a um ROI mais rápido para seus objetivos de negócios. Esses benefícios agora estão disponíveis para profissionais de MLOps usando o SageMaker Pipelines para automatizar o treinamento, o teste e a implantação de modelos de ML. Com o modo local, agora você pode iterar muito mais rapidamente enquanto desenvolve scripts para uso em um pipeline. Observe que as instâncias de pipeline locais não podem ser visualizadas ou executadas no Studio IDE; no entanto, opções de visualização adicionais para pipelines locais estarão disponíveis em breve.

O SDK do SageMaker fornece um propósito geral configuração do modo local que permite que os desenvolvedores executem e testem processadores e estimadores suportados em seu ambiente local. Você pode usar o treinamento no modo local com várias imagens de estrutura com suporte da AWS (TensorFlow, MXNet, Chainer, PyTorch e Scikit-Learn), bem como imagens que você mesmo fornece.

O SageMaker Pipelines, que cria um Gráfico Acíclico Dirigido (DAG) de etapas de fluxo de trabalho orquestradas, oferece suporte a muitas atividades que fazem parte do ciclo de vida de ML. No modo local, as seguintes etapas são suportadas:

• Processando etapas do trabalho – Uma experiência simplificada e gerenciada no SageMaker para executar cargas de trabalho de processamento de dados, como engenharia de recursos, validação de dados, avaliação de modelos e interpretação de modelos
• Etapas do trabalho de treinamento – Um processo iterativo que ensina um modelo a fazer previsões apresentando exemplos de um conjunto de dados de treinamento
• Jobs de ajuste de hiperparâmetros – Uma maneira automatizada de avaliar e selecionar os hiperparâmetros que produzem o modelo mais preciso
• Etapas de execução condicional – Uma etapa que fornece uma execução condicional de ramificações em um pipeline
• Etapa do modelo – Usando argumentos CreateModel, esta etapa pode criar um modelo para uso em etapas de transformação ou implantação posterior como um endpoint
• Transforme as etapas do trabalho – Um trabalho de transformação em lote que gera previsões de grandes conjuntos de dados e executa inferência quando um endpoint persistente não é necessário
• Etapas de falha – Uma etapa que interrompe uma execução de pipeline e marca a execução como com falha

Visão geral da solução

Nossa solução demonstra as etapas essenciais para criar e executar o SageMaker Pipelines no modo local, o que significa usar recursos locais de CPU, RAM e disco para carregar e executar as etapas do fluxo de trabalho. Seu ambiente local pode ser executado em um laptop, usando IDEs populares como VSCode ou PyCharm, ou pode ser hospedado pelo SageMaker usando instâncias clássicas de notebook.

O modo local permite que os cientistas de dados juntem etapas, que podem incluir trabalhos de processamento, treinamento e avaliação, e executem todo o fluxo de trabalho localmente. Quando terminar de testar localmente, você pode executar novamente o pipeline em um ambiente gerenciado pelo SageMaker substituindo o LocalPipelineSession objeto com PipelineSession, que traz consistência ao ciclo de vida de ML.

Para esta amostra de notebook, usamos um conjunto de dados padrão disponível publicamente, o Conjunto de dados Abalone de aprendizado de máquina UCI. O objetivo é treinar um modelo de ML para determinar a idade de um caracol abalone a partir de suas medidas físicas. No fundo, este é um problema de regressão.

Todo o código necessário para executar este exemplo de notebook está disponível no GitHub no exemplos de amazon-sagemaker repositório. Neste exemplo de notebook, cada etapa do fluxo de trabalho do pipeline é criada de forma independente e, em seguida, conectada para criar o pipeline. Criamos os seguintes passos:

• Etapa de processamento (engenharia de recursos)
• Etapa de treinamento (treinamento de modelo)
• Etapa de processamento (avaliação do modelo)
• Etapa de condição (precisão do modelo)
• Criar etapa de modelo (modelo)
• Etapa de transformação (transformação em lote)
• Etapa do modelo de registro (pacote de modelo)
• Falha na etapa (execução falhou)

O diagrama a seguir ilustra nosso pipeline.

Pré-requisitos

Para acompanhar este post, você precisa do seguinte:

Depois que esses pré-requisitos estiverem em vigor, você poderá executar o bloco de anotações de exemplo conforme descrito nas seções a seguir.

Construa seu pipeline

Neste exemplo de notebook, usamos Modo de script do SageMaker para a maioria dos processos de ML, o que significa que fornecemos o código Python real (scripts) para realizar a atividade e passar uma referência a esse código. O Modo Script oferece grande flexibilidade para controlar o comportamento no processamento do SageMaker, permitindo que você personalize seu código enquanto ainda aproveita os contêineres pré-construídos do SageMaker, como XGBoost ou Scikit-Learn. O código personalizado é gravado em um arquivo de script Python usando células que começam com o comando magic %%writefile, como o seguinte:

%%writefile code/evaluation.py

O ativador primário do modo local é o LocalPipelineSession objeto, que é instanciado do SDK do Python. Os segmentos de código a seguir mostram como criar um pipeline do SageMaker no modo local. Embora você possa configurar um caminho de dados local para muitas das etapas do pipeline local, o Amazon S3 é o local padrão para armazenar a saída de dados pela transformação. O novo LocalPipelineSession objeto é passado para o SDK do Python em muitas das chamadas da API de fluxo de trabalho do SageMaker descritas nesta postagem. Observe que você pode usar o local_pipeline_session variável para recuperar referências ao bucket padrão do S3 e o nome da região atual.

from sagemaker.workflow.pipeline_context import LocalPipelineSession

# Create a `LocalPipelineSession` object so that each 
# pipeline step will run locally
# To run this pipeline in the cloud, you must change 
# the `LocalPipelineSession()` to `PipelineSession()`
local_pipeline_session = LocalPipelineSession()
region = local_pipeline_session.boto_region_name

default_bucket = local_pipeline_session.default_bucket()
prefix = "sagemaker-pipelines-local-mode-example"

Antes de criarmos as etapas individuais do pipeline, definimos alguns parâmetros usados ​​pelo pipeline. Alguns desses parâmetros são literais de string, enquanto outros são criados como tipos enumerados especiais fornecidos pelo SDK. A tipagem enumerada garante que as configurações válidas sejam fornecidas ao pipeline, como esta, que é passada para o ConditionLessThanOrEqualTo passo mais para baixo:

mse_threshold = ParameterFloat(name="MseThreshold", default_value=7.0)

Para criar uma etapa de processamento de dados, que é usada aqui para realizar engenharia de recursos, usamos o SKLearnProcessor para carregar e transformar o conjunto de dados. Nós passamos o local_pipeline_session variável para o construtor de classe, que instrui a etapa de fluxo de trabalho a ser executada no modo local:

from sagemaker.sklearn.processing import SKLearnProcessor

framework_version = "1.0-1"

sklearn_processor = SKLearnProcessor(
    framework_version=framework_version,
    instance_type=instance_type,
    instance_count=processing_instance_count,
    base_job_name="sklearn-abalone-process",
    role=role,
    sagemaker_session=local_pipeline_session,
)

Em seguida, criamos nossa primeira etapa de pipeline real, uma ProcessingStep objeto, conforme importado do SDK do SageMaker. Os argumentos do processador são retornados de uma chamada para o SKLearnProcessor método run(). Essa etapa do fluxo de trabalho é combinada com outras etapas no final do notebook para indicar a ordem de operação no pipeline.

from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep

processor_args = sklearn_processor.run(
    inputs=[
        ProcessingInput(source=input_data, destination="/opt/ml/processing/input"),
    ],
    outputs=[
        ProcessingOutput(output_name="train", source="/opt/ml/processing/train"),
        ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"),
        ProcessingOutput(output_name="test", source="/opt/ml/processing/test"),
    ],
    code="code/preprocessing.py",
)

step_process = ProcessingStep(name="AbaloneProcess", step_args=processor_args)

Em seguida, fornecemos código para estabelecer uma etapa de treinamento instanciando primeiro um estimador padrão usando o SDK do SageMaker. Passamos o mesmo local_pipeline_session variável para o estimador, chamado xgb_train, como o sagemaker_session argumento. Como queremos treinar um modelo XGBoost, devemos gerar um URI de imagem válido especificando os seguintes parâmetros, incluindo o framework e vários parâmetros de versão:

from sagemaker.estimator import Estimator
from sagemaker.inputs import TrainingInput

model_path = f"s3://{default_bucket}/{prefix}/model"
image_uri = sagemaker.image_uris.retrieve(
    framework="xgboost",
    region=region,
    version="1.5-1",
    py_version="py3",
    instance_type=instance_type,
)

xgb_train = Estimator(
    image_uri=image_uri,
    entry_point="code/abalone.py",
    instance_type=instance_type,
    instance_count=training_instance_count,
    output_path=model_path,
    role=role,
    sagemaker_session=local_pipeline_session,
)

Opcionalmente, podemos chamar métodos estimadores adicionais, por exemplo set_hyperparameters(), para fornecer configurações de hiperparâmetro para o trabalho de treinamento. Agora que temos um estimador configurado, estamos prontos para criar a etapa de treinamento real. Mais uma vez, importamos o TrainingStep classe da biblioteca SageMaker SDK:

from sagemaker.workflow.steps import TrainingStep

step_train = TrainingStep(name="AbaloneTrain", step_args=train_args)

Em seguida, construímos outra etapa de processamento para realizar a avaliação do modelo. Isso é feito criando um ScriptProcessor instância e passando o local_pipeline_session objeto como parâmetro:

from sagemaker.processing import ScriptProcessor

script_eval = ScriptProcessor(
    image_uri=image_uri,
    command=["python3"],
    instance_type=instance_type,
    instance_count=processing_instance_count,
    base_job_name="script-abalone-eval",
    role=role,
    sagemaker_session=local_pipeline_session,
)

Para habilitar a implantação do modelo treinado, seja para um Ponto de extremidade em tempo real do SageMaker ou para uma transformação em lote, precisamos criar um Model objeto passando os artefatos de modelo, o URI de imagem apropriado e, opcionalmente, nosso código de inferência personalizado. Passamos então isso Model objetar a um ModelStep, que é adicionado ao pipeline local. Veja o seguinte código:

from sagemaker.model import Model

model = Model(
    image_uri=image_uri,
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    source_dir="code",
    entry_point="inference.py",
    role=role,
    sagemaker_session=local_pipeline_session,
)

from sagemaker.workflow.model_step import ModelStep

step_create_model = ModelStep(name="AbaloneCreateModel", 
    step_args=model.create(instance_type=instance_type)
)

Em seguida, criamos uma etapa de transformação em lote onde enviamos um conjunto de vetores de recursos e realizamos a inferência. Primeiro precisamos criar um Transformer objeto e passar o local_pipeline_session parâmetro para isso. Então criamos um TransformStep, passando os argumentos necessários e adicione isso à definição do pipeline:

from sagemaker.transformer import Transformer

transformer = Transformer(
    model_name=step_create_model.properties.ModelName,
    instance_type=instance_type,
    instance_count=transform_instance_count,
    output_path=f"s3://{default_bucket}/{prefix}/transform",
    sagemaker_session=local_pipeline_session,
)

from sagemaker.workflow.steps import TransformStep

transform_args = transformer.transform(transform_data, content_type="text/csv")

step_transform = TransformStep(name="AbaloneTransform", step_args=transform_args)

Por fim, queremos adicionar uma condição de ramificação ao fluxo de trabalho para que só executemos a transformação em lote se os resultados da avaliação do modelo atenderem aos nossos critérios. Podemos indicar essa condicional adicionando um ConditionStep com um tipo de condição específico, como ConditionLessThanOrEqualTo. Em seguida, enumeramos as etapas para as duas ramificações, essencialmente definindo as ramificações if/else ou true/false do pipeline. Os if_steps fornecidos no ConditionStep (step_create_model, etapa_transformação) são executados sempre que a condição for avaliada como True.

from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.functions import JsonGet

cond_lte = ConditionLessThanOrEqualTo(
    left=JsonGet(
        step_name=step_eval.name,
        property_file=evaluation_report,
        json_path="regression_metrics.mse.value",),
    right=mse_threshold,
)

step_cond = ConditionStep(
    name="AbaloneMSECond",
    conditions=[cond_lte],
    if_steps=[step_create_model, step_transform],
    else_steps=[step_fail],
)

O diagrama a seguir ilustra essa ramificação condicional e as etapas if/else associadas. Apenas uma ramificação é executada, com base no resultado da etapa de avaliação do modelo em comparação com a etapa de condição.

Agora que temos todas as nossas etapas definidas e as instâncias de classe subjacentes criadas, podemos combiná-las em um pipeline. Fornecemos alguns parâmetros e definimos crucialmente a ordem de operação simplesmente listando as etapas na ordem desejada. Observe que o TransformStep não é mostrado aqui porque é o destino da etapa condicional e foi fornecido como argumento de etapa para o ConditionalStep mais cedo.

from sagemaker.workflow.pipeline import Pipeline

pipeline_name = f"LocalModelPipeline"
pipeline = Pipeline(
    name=pipeline_name,
    parameters=[
        input_data,
        mse_threshold,
    ],
    steps=[step_process, step_train, step_eval, step_cond],
    sagemaker_session=local_pipeline_session,
)

Para executar o pipeline, você deve chamar dois métodos: pipeline.upsert(), que carrega o pipeline para o serviço subjacente e pipeline.start(), que inicia a execução do pipeline. Você pode usar vários outros métodos para interrogar o status de execução, listar as etapas do pipeline e muito mais. Como usamos a sessão de pipeline de modo local, todas essas etapas são executadas localmente em seu processador. A saída da célula abaixo do método start mostra a saída do pipeline:

pipeline.upsert(role_arn=role)
execution = pipeline.start()

Você deve ver uma mensagem na parte inferior da saída da célula semelhante à seguinte:

Pipeline execution d8c3e172-089e-4e7a-ad6d-6d76caf987b7 SUCCEEDED

Reverter para recursos gerenciados

Depois de confirmarmos que o pipeline é executado sem erros e estivermos satisfeitos com o fluxo e a forma do pipeline, podemos recriar o pipeline, mas com recursos gerenciados pelo SageMaker, e executá-lo novamente. A única alteração necessária é usar o PipelineSession objeto em vez de LocalPipelineSession:

da sagemaker.workflow.pipeline_context importar LocalPipelineSession
from sagemaker.workflow.pipeline_context import PipelineSession

local_pipeline_session = LocalPipelineSession()
pipeline_session = PipelineSession()

Isso informa o serviço para executar cada etapa que faz referência a esse objeto de sessão nos recursos gerenciados do SageMaker. Dada a pequena alteração, ilustramos apenas as alterações de código necessárias na célula de código a seguir, mas a mesma alteração precisaria ser implementada em cada célula usando o local_pipeline_session objeto. As alterações são, no entanto, idênticas em todas as células porque estamos apenas substituindo o local_pipeline_session objeto com o pipeline_session objeto.

from sagemaker.sklearn.processing import SKLearnProcessor

framework_version = "1.0-1"

sklearn_processor = SKLearnProcessor(
    framework_version=framework_version,
    instance_type=instance_type,
    instance_count=processing_instance_count,
    base_job_name="sklearn-abalone-process",
    role=role,
    sagemaker_session=pipeline_session,  # non-local session
)

Após a substituição do objeto de sessão local em todos os lugares, recriamos o pipeline e o executamos com recursos gerenciados pelo SageMaker:

from sagemaker.workflow.pipeline import Pipeline

pipeline_name = f"LocalModelPipeline"
pipeline = Pipeline(
    name=pipeline_name,
    parameters=[
        input_data,
        mse_threshold,
    ],
    steps=[step_process, step_train, step_eval, step_cond],
    sagemaker_session=pipeline_session, # non-local session
)

pipeline.upsert(role_arn=role)
execution = pipeline.start()

limpar

Se quiser manter o ambiente do Studio organizado, você pode usar os métodos a seguir para excluir o pipeline do SageMaker e o modelo. O código completo pode ser encontrado no exemplo caderno.

# delete models 
sm_client = boto3.client("sagemaker")
model_prefix="AbaloneCreateModel"
delete_models(sm_client, model_prefix)

# delete managed pipeline
pipeline_to_delete = 'SM-Managed-Pipeline'
delete_sagemaker_pipeline(sm_client, pipeline_to_delete)

Conclusão

Até recentemente, você podia usar o recurso de modo local do SageMaker Processing e do SageMaker Training para iterar seus scripts de processamento e treinamento localmente, antes de executá-los em todos os dados com recursos gerenciados pelo SageMaker. Com o novo recurso de modo local do SageMaker Pipelines, os profissionais de ML agora podem aplicar o mesmo método ao iterar em seus pipelines de ML, unindo os diferentes fluxos de trabalho de ML. Quando o pipeline estiver pronto para produção, executá-lo com recursos gerenciados pelo SageMaker requer apenas algumas linhas de alterações de código. Isso reduz o tempo de execução do pipeline durante o desenvolvimento, levando a um desenvolvimento de pipeline mais rápido com ciclos de desenvolvimento mais rápidos, ao mesmo tempo em que reduz o custo dos recursos gerenciados pelo SageMaker.

Para saber mais, visite Pipelines Amazon SageMaker or Use o SageMaker Pipelines para executar seus trabalhos localmente.

Sobre os autores

Paulo Hargis concentrou seus esforços em aprendizado de máquina em várias empresas, incluindo AWS, Amazon e Hortonworks. Ele gosta de criar soluções de tecnologia e ensinar as pessoas a tirar o máximo proveito delas. Antes de sua função na AWS, ele foi arquiteto-chefe da Amazon Exports and Expansions, ajudando a amazon.com a melhorar a experiência de compradores internacionais. Paul gosta de ajudar os clientes a expandir suas iniciativas de aprendizado de máquina para resolver problemas do mundo real.

Niklas Palma é Arquiteto de Soluções na AWS em Estocolmo, Suécia, onde ajuda clientes em todos os países nórdicos a ter sucesso na nuvem. Ele é particularmente apaixonado por tecnologias sem servidor, juntamente com IoT e aprendizado de máquina. Fora do trabalho, Niklas é um ávido esquiador de cross-country e snowboarder, bem como um mestre na caldeira de ovos.

Kirit Thadaka é um arquiteto de soluções de ML que trabalha na equipe SageMaker Service SA. Antes de ingressar na AWS, Kirit trabalhou em startups de IA em estágio inicial, seguido por algum tempo de consultoria em várias funções em pesquisa de IA, MLOps e liderança técnica.