As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.
# Lift-and-shift Código Python com o decorador @step
O decorador `@step` é um atributo que converte seu código local de machine learning (ML) em uma ou mais etapas de pipeline. Você pode escrever sua função de ML da mesma forma que faria em qualquer projeto de ML. Depois de testada localmente ou como um trabalho de treinamento usando o `@remote` decorador, você pode converter a função em uma etapa do pipeline de SageMaker IA adicionando um `@step` decorador. Em seguida, você pode passar o resultado da chamada da função decorada por `@step` como uma etapa para o Pipelines, para criar e executar um pipeline. Você também pode encadear uma série de funções com o decorador `@step` para criar um pipeline de gráfico acíclico direcionado (DAG) de várias etapas.
A configuração para usar o decorador `@step` é a mesma configuração para usar o decorador `@remote`. Você pode consultar a documentação da função remota para obter detalhes sobre como [configurar o ambiente](https://docs.aws.amazon.com/sagemaker/latest/dg/train-remote-decorator.html#train-remote-decorator-env) e [usar um arquivo de configuração](https://docs.aws.amazon.com/sagemaker/latest/dg/train-remote-decorator-config.html) para definir padrões. Para obter mais informações sobre o decorador `@step`, consulte [sagemaker.workflow.function\$1step.step](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.function_step.step).
Para ver cadernos de exemplo que demonstram o uso do decorador `@step`, consulte [cadernos de exemplo do decorador @step](https://github.com/aws/amazon-sagemaker-examples/tree/main/sagemaker-pipelines/step-decorator).
As seções a seguir explicam como você pode anotar seu código de ML local com um decorador `@step` para criar uma etapa, criar e executar um pipeline usando a etapa e personalizar a experiência para seu caso de uso.
**Topics**
+ [Criar um pipeline com funções decoradas por `@step`](pipelines-step-decorator-create-pipeline.md)
+ [Execute um pipeline](pipelines-step-decorator-run-pipeline.md)
+ [Configure seu pipeline](pipelines-step-decorator-cfg-pipeline.md)
+ [Práticas recomendadas](pipelines-step-decorator-best.md)
+ [Limitações](pipelines-step-decorator-limit.md)
# Criar um pipeline com funções decoradas por `@step`
Você pode criar um pipeline convertendo funções do Python em etapas de pipeline usando o decorador `@step`, criando dependências entre essas funções para criar um gráfico de pipeline (ou gráfico acíclico direcionado (DAG)) e passando os nós da folha desse gráfico como uma lista de etapas para o pipeline. As seções a seguir explicam esse procedimento detalhadamente com exemplos.
**Topics**
+ [Converter uma função em uma etapa](#pipelines-step-decorator-run-pipeline-convert)
+ [Crie dependências entre as etapas](#pipelines-step-decorator-run-pipeline-link)
+ [Use `ConditionStep` com etapas decoradas por `@step`](#pipelines-step-decorator-condition)
+ [Definir um pipeline usando a saída `DelayedReturn` das etapas](#pipelines-step-define-delayed)
+ [Criar um pipeline](#pipelines-step-decorator-pipeline-create)
## Converter uma função em uma etapa
Para criar uma etapa usando o decorador `@step`, anote a função com `@step`. O exemplo a seguir mostra uma função decorada por `@step` que pré-processa os dados.
```
from sagemaker.workflow.function_step import step
@step
def preprocess(raw_data):
df = pandas.read_csv(raw_data)
...
return procesed_dataframe
step_process_result = preprocess(raw_data)
```
Quando você invoca uma função `@step` -decorada, a SageMaker IA retorna uma `DelayedReturn` instância em vez de executar a função. Uma instância `DelayedReturn` é um proxy para o retorno real da função. A instância `DelayedReturn` pode ser passada para outra função como argumento ou diretamente para uma instância do pipeline como uma etapa. Para obter informações sobre a `DelayedReturn` classe, consulte [sagemaker.workflow.function\$1step. DelayedReturn](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.function_step.DelayedReturn).
## Crie dependências entre as etapas
Ao criar uma dependência entre duas etapas, você cria uma conexão entre as etapas no gráfico do pipeline. As seções a seguir apresentam várias maneiras de criar uma dependência entre as etapas do pipeline.
### Dependências de dados por meio de argumentos de entrada
Passar a saída `DelayedReturn` de uma função como entrada para outra função cria automaticamente uma dependência de dados no DAG do pipeline. No exemplo a seguir, passar a saída `DelayedReturn` da função `preprocess` para a função `train` cria uma dependência entre `preprocess` e `train`.
```
from sagemaker.workflow.function_step import step
@step
def preprocess(raw_data):
df = pandas.read_csv(raw_data)
...
return procesed_dataframe
@step
def train(training_data):
...
return trained_model
step_process_result = preprocess(raw_data)
step_train_result = train(step_process_result)
```
O exemplo anterior define uma função de treinamento que é decorada com`@step`. Quando essa função é invocada, ela recebe a saída `DelayedReturn` da etapa do pipeline de pré-processamento como entrada. A invocação da função de treinamento retorna outra instância `DelayedReturn`. Essa instância contém as informações sobre todas as etapas anteriores definidas nessa função (ou seja, a etapa `preprocess` no exemplo) que formam o DAG do pipeline.
No exemplo anterior, a função `preprocess` retorna um único valor. Para tipos de retorno mais complexos, como listas ou tuplas, consulte [Limitações](pipelines-step-decorator-limit.md).
### Definir dependências personalizadas
No exemplo anterior, a função `train` recebeu a saída `DelayedReturn` de `preprocess` e criou uma dependência. Se você quiser definir a dependência explicitamente sem passar a saída da etapa anterior, use a função `add_depends_on` com a etapa. Você pode usar a função `get_step()` para recuperar a etapa subjacente de sua instância `DelayedReturn` e, em seguida, chamar `add_depends_on`\$1on com a dependência como entrada. Para ver a definição da função `get_step()`, consulte [sagemaker.workflow.step\$1outputs.get\$1step](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.step_outputs.get_step). O exemplo a seguir mostra como criar uma dependência entre `preprocess` e `train` usando `get_step()` e `add_depends_on()`.
```
from sagemaker.workflow.step_outputs import get_step
@step
def preprocess(raw_data):
df = pandas.read_csv(raw_data)
...
processed_data = ..
return s3.upload(processed_data)
@step
def train():
training_data = s3.download(....)
...
return trained_model
step_process_result = preprocess(raw_data)
step_train_result = train()
get_step(step_train_result).add_depends_on([step_process_result])
```
### Transmita dados de e para uma função decorada por `@step` para uma etapa tradicional de pipeline
Você pode criar um pipeline que inclua uma etapa decorada por `@step` e uma etapa tradicional de pipeline para transmitir dados entre elas. Por exemplo, você pode usar `ProcessingStep` para processar os dados e passar o resultado para a função de treinamento decorada por `@step`. No exemplo a seguir, uma etapa de treinamento decorada por `@step` faz referência à saída de uma etapa de processamento.
```
# Define processing step
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep
sklearn_processor = SKLearnProcessor(
framework_version='1.2-1',
role='arn:aws:iam::123456789012:role/SagemakerExecutionRole',
instance_type='ml.m5.large',
instance_count='1',
)
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")
]
process_step = ProcessingStep(
name="MyProcessStep",
step_args=sklearn_processor.run(inputs=inputs, outputs=outputs,code='preprocessing.py'),
)
```
```
# Define a @step-decorated train step which references the
# output of a processing step
@step
def train(train_data_path, test_data_path):
...
return trained_model
step_train_result = train(
process_step.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
process_step.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri,
)
```
## Use `ConditionStep` com etapas decoradas por `@step`
Os pipelines oferecem apoio a uma classe `ConditionStep` que avalia os resultados das etapas anteriores para decidir qual ação tomar no pipeline. Você também pode usar `ConditionStep` com uma etapa decorada por `@step`. Para usar a saída de qualquer etapa decorada por `@step` com `ConditionStep`, insira a saída dessa etapa como argumento para `ConditionStep`. No exemplo a seguir, a etapa de condição recebe a saída da etapa de avaliação de modelo decorada por `@step`.
```
# Define steps
@step(name="evaluate")
def evaluate_model():
# code to evaluate the model
return {
"rmse":rmse_value
}
@step(name="register")
def register_model():
# code to register the model
...
```
```
# Define ConditionStep
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.fail_step import FailStep
conditionally_register = ConditionStep(
name="conditional_register",
conditions=[
ConditionGreaterThanOrEqualTo(
# Output of the evaluate step must be json serializable
left=evaluate_model()["rmse"], #
right=5,
)
],
if_steps=[FailStep(name="Fail", error_message="Model performance is not good enough")],
else_steps=[register_model()],
)
```
## Definir um pipeline usando a saída `DelayedReturn` das etapas
Você define um pipeline da mesma forma, independentemente de usar ou não um decorador `@step`. Ao passar uma instância `DelayedReturn` para seu pipeline, você não precisa passar uma lista completa de etapas para criar o pipeline. O SDK infere automaticamente as etapas anteriores com base nas dependências que você define. Todas as etapas anteriores dos objetos `Step` que você passou para o pipeline ou objetos `DelayedReturn` são incluídas no gráfico do pipeline. No exemplo a seguir, o pipeline recebe o objeto `DelayedReturn` da função `train`. SageMaker A IA adiciona a `preprocess` etapa, como etapa anterior`train`, ao gráfico do pipeline.
```
from sagemaker.workflow.pipeline import Pipeline
pipeline = Pipeline(
name="",
steps=[step_train_result],
sagemaker_session=,
)
```
Se não houver dados ou dependências personalizadas entre as etapas e você executar várias etapas em paralelo, o gráfico do pipeline terá mais de um nó de folha. Passe todos esses nós de folha em uma lista para o argumento `steps` em sua definição de pipeline, conforme mostrado no seguinte exemplo:
```
@step
def process1():
...
return data
@step
def process2():
...
return data
step_process1_result = process1()
step_process2_result = process2()
pipeline = Pipeline(
name="",
steps=[step_process1_result, step_process2_result],
sagemaker_session=sagemaker-session,
)
```
Quando o pipeline é executado, as duas etapas são executadas paralelamente.
Você só passa os nós de folha do gráfico para o pipeline porque os nós de folha contêm informações sobre todas as etapas anteriores definidas por meio de dados ou dependências personalizadas. Ao compilar o pipeline, a SageMaker IA também infere todas as etapas subsequentes que formam o gráfico do pipeline e adiciona cada uma delas como uma etapa separada ao pipeline.
## Criar um pipeline
Crie um pipeline chamando `pipeline.create()`, como mostrado no trecho a seguir. Para obter detalhes sobre `create()`, consulte [sagemaker.workflow.pipeline.Pipeline.create](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.pipeline.Pipeline.create).
```
role = "pipeline-role"
pipeline.create(role)
```
Quando você liga`pipeline.create()`, a SageMaker IA compila todas as etapas definidas como parte da instância do pipeline. SageMaker A IA carrega a função serializada, os argumentos e todos os outros artefatos relacionados à etapa para o Amazon S3.
Os dados residem no bucket do S3 de acordo com a seguinte estrutura:
```
s3_root_uri/
pipeline_name/
sm_rf_user_ws/
workspace.zip # archive of the current working directory (workdir)
step_name/
timestamp/
arguments/ # serialized function arguments
function/ # serialized function
pre_train_dependencies/ # any dependencies and pre_execution scripts provided for the step
execution_id/
step_name/
results # returned output from the serialized function including the model
```
`s3_root_uri`é definido no arquivo de configuração do SageMaker AI e se aplica a todo o pipeline. Se indefinido, o bucket de SageMaker IA padrão será usado.
**nota**
Toda vez que a SageMaker IA compila um pipeline, a SageMaker IA salva as funções, argumentos e dependências serializados das etapas em uma pasta com a data e hora atual. Isso ocorre toda vez que você executa `pipeline.create()`, `pipeline.update()`, `pipeline.upsert()` ou `pipeline.definition()`.
# Execute um pipeline
A página a seguir descreve como executar um pipeline com o Amazon SageMaker Pipelines, com recursos de SageMaker IA ou localmente.
Inicie uma nova execução de pipeline com a `pipeline.start()` função, como você faria com uma execução de pipeline de SageMaker IA tradicional. Para obter informações sobre a função `start()`, consulte [sagemaker.workflow.pipeline.Pipeline.start](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.pipeline.Pipeline.start).
**nota**
Uma etapa definida usando o decorador `@step` é executada como um trabalho de treinamento. Portanto, esteja ciente das seguintes limitações:
Limites de instância e limites de tarefas de treinamento em suas contas. Atualize seus limites adequadamente para evitar problemas de limitação de recursos ou controle de utilização.
Os custos monetários associados a cada execução de uma etapa de treinamento no pipeline. Para obter mais detalhes, consulte [Amazon SageMaker Pricing](https://aws.amazon.com/sagemaker/pricing/).
## Recupere resultados de uma execução de pipeline localmente
Para ver o resultado de qualquer etapa da execução de um pipeline, use [execution.result()](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.pipeline._PipelineExecution.result ), conforme mostrado no trecho a seguir:
```
execution = pipeline.start()
execution.result(step_name="train")
```
**nota**
O Pipelines não oferece apoio ao `execution.result()` no modo local.
Você só poderá recuperar resultados de uma etapa de cada vez. Se o nome da etapa foi gerado pela SageMaker IA, você pode recuperar o nome da etapa chamando da `list_steps` seguinte forma:
```
execution.list_step()
```
## Execute um pipeline localmente
Você pode executar um pipeline com etapas decoradas por `@step` localmente, como faria com as etapas tradicionais do pipeline. Para obter detalhes sobre a execução de pipeline no modo local, consulte [Executar pipelines usando o modo local](pipelines-local-mode.md). Para usar o modo local, forneça um `LocalPipelineSession` em vez de `SageMakerSession` para sua definição de pipeline, conforme mostrado no seguinte exemplo:
```
from sagemaker.workflow.function_step import step
from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.pipeline_context import LocalPipelineSession
@step
def train():
training_data = s3.download(....)
...
return trained_model
step_train_result = train()
local_pipeline_session = LocalPipelineSession()
local_pipeline = Pipeline(
name="",
steps=[step_train_result],
sagemaker_session=local_pipeline_session # needed for local mode
)
local_pipeline.create(role_arn="role_arn")
# pipeline runs locally
execution = local_pipeline.start()
```
# Configure seu pipeline
É recomendável usar o arquivo de configuração do SageMaker AI para definir os padrões do pipeline. Para obter informações sobre o arquivo de configuração do SageMaker AI, consulte [Como configurar e usar padrões com o SDK do Python SageMaker ](https://sagemaker.readthedocs.io/en/stable/overview.html#configuring-and-using-defaults-with-the-sagemaker-python-sdk). Qualquer configuração adicionada ao arquivo de configuração se aplica a todas as etapas do pipeline. Se você quiser substituir as opções de alguma etapa, forneça novos valores nos argumentos do decorador `@step`. O tópico a seguir descreve como configurar um arquivo de configuração.
A configuração do decorador `@step` no arquivo de configuração é idêntica à configuração do decorador `@remote`. Para configurar o ARN da função do pipeline e as tags do pipeline no arquivo de configuração, use a seção `Pipeline` mostrada no seguinte trecho:
```
SchemaVersion: '1.0'
SageMaker:
Pipeline:
RoleArn: 'arn:aws:iam::555555555555:role/IMRole'
Tags:
- Key: 'tag_key'
Value: 'tag_value'
```
Para a maioria dos padrões que você pode definir no arquivo de configuração, você também pode substituir passando novos valores para o decorador `@step`. Por exemplo, é possível substituir o tipo de instância definido no arquivo de configuração da etapa de pré-processamento, conforme mostrado no seguinte exemplo:
```
@step(instance_type="ml.m5.large")
def preprocess(raw_data):
df = pandas.read_csv(raw_data)
...
return procesed_dataframe
```
Alguns argumentos não fazem parte da lista de parâmetros do `@step` decorador. Eles só podem ser configurados para todo o pipeline por meio do arquivo de configuração do SageMaker AI. Eles estão listados a seguir:
+ `sagemaker_session`(`sagemaker.session.Session`): A sessão de SageMaker IA subjacente à qual a SageMaker IA delega chamadas de serviço. Se não for especificado, uma sessão será criada usando a seguinte configuração padrão:
```
SageMaker:
PythonSDK:
Modules:
Session:
DefaultS3Bucket: 'default_s3_bucket'
DefaultS3ObjectKeyPrefix: 'key_prefix'
```
+ `custom_file_filter` (`CustomFileFilter)`: um objeto `CustomFileFilter` que especifica os diretórios e arquivos locais a serem incluídos na etapa do pipeline. Se não for especificado, esse valor terá o padrão `None`. Para `custom_file_filter` entrar em vigor, você deve definir `IncludeLocalWorkdir` como `True`. O exemplo a seguir mostra uma configuração que ignora todos os arquivos do caderno e os arquivos e diretórios chamados `data`.
```
SchemaVersion: '1.0'
SageMaker:
PythonSDK:
Modules:
RemoteFunction:
IncludeLocalWorkDir: true
CustomFileFilter:
IgnoreNamePatterns: # files or directories to ignore
- "*.ipynb" # all notebook files
- "data" # folder or file named "data"
```
Para obter mais detalhes sobre como usar `IncludeLocalWorkdir` com `CustomFileFilter`, consulte [Como usar o código modular com o decorador @remote](train-remote-decorator-modular.md).
+ `s3_root_uri (str)`: a pasta raiz do Amazon S3 para a qual a SageMaker IA carrega os arquivos de código e os dados. Se não for especificado, o bucket de SageMaker IA padrão será usado.
+ `s3_kms_key (str)`: a chave usada para criptografar os dados de entrada e saída. Você só pode configurar esse argumento no arquivo de configuração do SageMaker AI e o argumento se aplica a todas as etapas definidas no pipeline. Se não especificado, o valor padrão é `None`. Veja o seguinte trecho para ver um exemplo de configuração de chave do KMS do S3:
```
SchemaVersion: '1.0'
SageMaker:
PythonSDK:
Modules:
RemoteFunction:
S3KmsKeyId: 's3kmskeyid'
S3RootUri: 's3://amzn-s3-demo-bucket/my-project
```
# Práticas recomendadas
As seções a seguir sugerem as práticas recomendadas a serem seguidas ao usar o decorador `@step` nas etapas do pipeline.
## Use grupos de aquecimento
Para uma execução mais rápida das etapas de pipeline, use a funcionalidade de grupos de aquecimento fornecida para trabalhos de treinamento. Você pode ativar a funcionalidade de grupos de aquecimento fornecendo o argumento `keep_alive_period_in_seconds` ao decorador `@step`, conforme demonstrado no seguinte trecho:
```
@step(
keep_alive_period_in_seconds=900
)
```
Para obter mais informações sobre grupos quentes, consulte [SageMaker Piscinas aquecidas gerenciadas por IA](train-warm-pools.md).
## Proteja seu diretório
É recomendável usar módulos de código ao usar o decorador `@step`. Coloque o módulo `pipeline.py`, no qual você invoca as funções de etapa e define o pipeline, na raiz do espaço de trabalho. A estrutura recomendada é mostrada a seguir:
```
.
├── config.yaml # the configuration file that define the infra settings
├── requirements.txt # dependencies
├── pipeline.py # invoke @step-decorated functions and define the pipeline here
├── steps/
| ├── processing.py
| ├── train.py
├── data/
├── test/
```
# Limitações
As seções a seguir descrevem as limitações que você deve conhecer ao usar o decorador `@step` nas etapas do pipeline.
## Limitações de argumentos de função
Quando você passa um argumento de entrada para a função decorada por `@step`, as seguintes limitações se aplicam:
+ Você pode passar `DelayedReturn`, `Properties` (de etapas de outros tipos), `Parameter` e objetos `ExecutionVariable` para funções decoradas por `@step` como argumentos. Mas as funções decoradas por `@step` não oferecem apoio a `JsonGet` e objetos `Join` como argumentos.
+ Você não pode acessar diretamente uma variável de pipeline a partir de uma função `@step`. O seguinte exemplo produz um erro:
```
param = ParameterInteger(name="", default_value=10)
@step
def func():
print(param)
func() # this raises a SerializationError
```
+ Você não pode aninhar uma variável de pipeline em outro objeto e passá-la para uma função `@step`. O seguinte exemplo produz um erro:
```
param = ParameterInteger(name="", default_value=10)
@step
def func(arg):
print(arg)
func(arg=(param,)) # this raises a SerializationError because param is nested in a tuple
```
+ Como as entradas e saídas de uma função são serializadas, há restrições quanto ao tipo de dados que pode ser passado como entrada ou saída de uma função. Consulte a seção *Serialização e desserialização de dados* de [Invocar uma função remota](train-remote-decorator-invocation.md) para obter mais detalhes. As mesmas restrições se aplicam às funções decoradas por `@step`.
+ Qualquer objeto que tenha um cliente boto não pode ser serializado, portanto, você não pode passar esses objetos como entrada ou saída de uma função decorada por `@step`. Por exemplo, classes de cliente do SDK do SageMaker Python`Estimator`, como, e`Predictor`, não `Processor` podem ser serializadas.
## Importações de funções
Você deve importar as bibliotecas exigidas pela etapa dentro da função, ao invés de fora. Se você importá-las em âmbito global, corre o risco de uma colisão de importação ao serializar a função. Por exemplo, `sklearn.pipeline.Pipeline` pode ser substituído por `sagemaker.workflow.pipeline.Pipeline`.
## Como referenciar membros filhos do valor de retorno da função
Se você fizer referência a membros filhos do valor de retorno de uma função decorada por `@step`, as seguintes limitações se aplicam:
+ Você pode referenciar os membros secundários com `[]` se o objeto `DelayedReturn` representar uma tupla, lista ou dicionário, conforme mostrado no seguinte exemplo:
```
delayed_return[0]
delayed_return["a_key"]
delayed_return[1]["a_key"]
```
+ Você não pode descompactar uma saída de tupla ou lista porque não é possível saber o comprimento exato da tupla ou lista subjacente quando você invoca a função. O seguinte exemplo produz um erro:
```
a, b, c = func() # this raises ValueError
```
+ Você não pode repetir um objeto `DelayedReturn`. Veja a seguir um exemplo com erro:
```
for item in func(): # this raises a NotImplementedError
```
+ Você não pode referenciar membros secundários arbitrários com “`.`”. O seguinte exemplo produz um erro:
```
delayed_return.a_child # raises AttributeError
```
## Atributos de pipeline existentes que não são compatíveis
Você não pode usar o decorador `@step` com os seguintes atributos de pipeline:
+ [Cache de etapas de pipeline](https://docs.aws.amazon.com/sagemaker/latest/dg/pipelines-caching.html)
+ [Arquivo de propriedades](https://docs.aws.amazon.com/sagemaker/latest/dg/build-and-manage-propertyfile.html#build-and-manage-propertyfile-property)