为 AWS Step Functions SageMaker 创建自定义 Docker 容器镜像并将其用于模型训练

设置 Amazon ECR 并新建私有注册表。

如果您尚未设置 Amazon ECR，请按照Amazon ECR 用户指南中的设置 Amazon ECR进行操作。每个 Amazon Web Services account 都提供有原定设置的私有 Amazon ECR 注册表。

创建 Amazon ECR 私有存储库。

请按照Amazon ECR 用户指南中的创建私有存储库进行操作。

注意：您创建的存储库是存储自定义 Docker 容器映像的位置。

创建一个 Dockerfile，其中包含运行 SageMaker 处理作业所需的规范。

通过配置 Dockerfile 来创建包含运行 SageMaker 处理作业所需的规格的 Dockerfile。有关说明，请参阅《Amazon SageMaker 开发者指南》中的调整自己的训练容器。

有关 Dockerfiles 的更多信息，请参阅 Doc ker 文档中的 DockerFile 参考。

用于创建 DockerFile 的 Jupyter 笔记本代码单元格示例

单元格 1

# Make docker folder
!mkdir -p docker

单元格 2

%%writefile docker/Dockerfile

FROM python:3.7-slim-buster

RUN pip3 install pandas==0.25.3 scikit-learn==0.21.3
ENV PYTHONUNBUFFERED=TRUE

ENTRYPOINT ["python3"]

构建您的 Docker 容器映像并将其推送至 Amazon ECR。

有关更多信息，请参阅在上构建自己的算法容器中的构建和注册容器 GitHub。

用于构建和注册 Docker 映像的 Jupyter 笔记本代码单元格示例

重要提示：运行以下单元格之前，请确保已创建一个 Dockerfile 并将其存储在名为docker的目录中。此外，请确保您已创建一个 Amazon ECR 存储库，并将第一个单元格中的ecr_repository 值替换为存储库名称。

单元格 1

import boto3
tag = ':latest'
account_id = boto3.client('sts').get_caller_identity().get('Account')
region = boto3.Session().region_name
ecr_repository = 'byoc'

image_uri = '{}.dkr.ecr.{}.amazonaws.com/{}'.format(account_id, region, ecr_repository + tag)

单元格 2

# Build docker image
!docker build -t $image_uri docker

单元格 3

# Authenticate to ECR
!aws ecr get-login-password --region {region} | docker login --username AWS --password-stdin {account_id}.dkr.ecr.{region}.amazonaws.com

单元格 4

# Push docker image
!docker push $image_uri

注意：您必须对私有注册表的 Docker 客户端进行身份验证，这样才能使用 docker push和 docker pull命令。这些命令将图像推送和拉出注册表中存储库。

创建包含自定义处理和模型训练逻辑的 Python 脚本。

编写将在数据处理脚本中运行的自定义处理逻辑。然后，将其另存为名为training.py的 Python 脚本。

有关更多信息，请参阅在开启 SageMaker 脚本模式的情况下自带模型 GitHub。

含自定义处理和模型训练逻辑的 Python 脚本示例

%%writefile training.py
from numpy import empty
import pandas as pd
import os
from sklearn import datasets, svm
from joblib import dump, load


if __name__ == '__main__':
    digits = datasets.load_digits()
    #create classifier object
    clf = svm.SVC(gamma=0.001, C=100.)
    
    #fit the model
    clf.fit(digits.data[:-1], digits.target[:-1])
    
    #model output in binary format
    output_path = os.path.join('/opt/ml/processing/model', "model.joblib")
    dump(clf, output_path)

创建一个 Step Functions 工作流程，其中包含您的 SageMaker 处理作业。

安装并导入AWS Step Functions Data Science SDK，然后将training.py 文件上传至 Amazon S3。然后，使用 Amaz SageMaker on Python 开发工具包在 Step Functions 中定义处理步骤。

重要提示：请确保您已在 Amazon Web Services account 中为 Step Functions 创建 IAM 执行角色。

要上传至 Amazon S3 的环境设置和自定义训练脚本示例

!pip install stepfunctions

import boto3
import stepfunctions
import sagemaker
import datetime

from stepfunctions import steps
from stepfunctions.inputs import ExecutionInput
from stepfunctions.steps import (
    Chain
)
from stepfunctions.workflow import Workflow
from sagemaker.processing import ScriptProcessor, ProcessingInput, ProcessingOutput

sagemaker_session = sagemaker.Session()
bucket = sagemaker_session.default_bucket() 
role = sagemaker.get_execution_role()
prefix = 'byoc-training-model'

# See prerequisites section to create this role
workflow_execution_role = f"arn:aws:iam::{account_id}:role/AmazonSageMaker-StepFunctionsWorkflowExecutionRole"

execution_input = ExecutionInput(
    schema={
        "PreprocessingJobName": str})


input_code = sagemaker_session.upload_data(
    "training.py",
    bucket=bucket,
    key_prefix="preprocessing.py",
)

使用自定义 Amazon ECR 图像和 Python 脚本的 SageMaker 处理步骤定义示例

注意：请务必使用 execution_input参数指定作业名称。每次运行作业时，参数值必须是唯一的。此外，training.py 文件的代码作为input 参数传递至ProcessingStep，这意味着它将被复制到容器中。ProcessingInput 代码的目标与container_entrypoint内部的第二个参数相同。

script_processor = ScriptProcessor(command=['python3'],
                image_uri=image_uri,
                role=role,
                instance_count=1,
                instance_type='ml.m5.xlarge')


processing_step = steps.ProcessingStep(
    "training-step",
    processor=script_processor,
    job_name=execution_input["PreprocessingJobName"],
    inputs=[
        ProcessingInput(
            source=input_code,
            destination="/opt/ml/processing/input/code",
            input_name="code",
        ),
    ],
    outputs=[
        ProcessingOutput(
            source='/opt/ml/processing/model', 
            destination="s3://{}/{}".format(bucket, prefix), 
            output_name='byoc-example')
    ],
    container_entrypoint=["python3", "/opt/ml/processing/input/code/training.py"],
)

运行处理作业的 Step Functions 工作 SageMaker 流程示例

注意：此示例工作流仅包括 SageMaker 处理作业步骤，不包括完整的 Step Functions 工作流程。有关完整的工作流程示例，请参阅 AWS Step Functions 数据科学软件开发工具包文档 SageMaker中的示例笔记本。

workflow_graph = Chain([processing_step])

workflow = Workflow(
    name="ProcessingWorkflow",
    definition=workflow_graph,
    role=workflow_execution_role
)

workflow.create()
# Execute workflow
execution = workflow.execute(
    inputs={
        "PreprocessingJobName": str(datetime.datetime.now().strftime("%Y%m%d%H%M-%SS")),  # Each pre processing job (SageMaker processing job) requires a unique name,
    }
)
execution_output = execution.get_output(wait=True)