使用 Amazon SageMaker 助力 AI 内容生成

{"value":"\n\n### **场景简介**\n\n随着深度学习领域的不断发展，很多生成类模型的效果已经达到“准商用”的水平，因此，也有很多客户选择在这类模型上进行调优和封装并作为应用提供给终端用户，从而实现算法能力产品化。\n\n\n\n本文中，我们将为大家展示如何用利用 Amazon SageMaker Pipeline 构建包括 **数据预处理、模型训练、模型推理以及任务状态通知的自动化工作流**。在本文中我们也将为大家展示主要部分的技术实现，您可以参考**完整代码**（[https://github.com/DearAmber/SageMakerPipelineStyleGAN2]()）。\n\n### **模型介绍: StyleGAN2**\n\nStyleGAN 是目前最先进的高分辨率图像合成方法之一，它生成的人脸照片一度被认为“逼真到吓人”。在2021年初，英伟达 (NVIDIA) 的研究人员发布并开源了升级版本：StyleGAN2 。该版本重点修复了特征伪影问题，并在进一步提高生成图像质量的同时降低了训练时间和训练集大小。\n\n\n\n自面世以来，StyleGAN2 受到了众多智能视觉爱好者的欢迎，其实践除了人脸生成以外还包括宠物生成、动漫生成、场景生成等等。广泛的应用场景也引起了艺术设计领域从业者的兴趣，他们希望借助此模型来丰富他们现有平台的功能，利用 AI 来赋能内容生成。\n\n\n\nStyleGAN2：\n\n[https://github.com/NVlabs/stylegan2-ada-pytorch]()\n\n### **数据介绍: Anime Face Dataset**\n\nAnime Face Dataset 是一个来自于 Kaggle 平台的开源数据集。里面包含63,632张卡通头像图片可用于做各种各样的深度学习实验。\n\n\n\n\nAnime Face Dataset：\n[https://www.kaggle.com/datasets/splcher/animefacedataset]()\n\n### **工程架构参考**\n\n如下图所示，我们不妨模拟为平台客户封装 AI 增值服务的场景。在该场景中，平台客户的终端用户有可能是原画师或者设计师，他们仅需要在客户的前端网页上上传标注好的图像素材，平台客户则仅需要在后台触发预设好的 SageMaker Pipeline 任务，SageMaker Pipeline 将会自动获取 S3 中的训练数据并开启整个工作流，并按需回传训练的中间结果，最终的模型文件和推理结果。在推理完成之后，SageMaker Pipeline 会通过 Amazon SNS 通知平台终端用户。\n\n\n\n\n### **部分实验结果**\n\n本文的实验框架环境为 Python3.8，PyTorch1.8.1+cu111。同时选用了单个 ml.p3.2xlarge 作为训练计算资源。\n\n\n\n下图为完整训练后部分生成样本，仅供参考。\n\n\n\n完整的训练将需要持续72小时左右，请参考以下超参：\n\n```\nhyperparameters = {\"data\":\"/opt/ml/input/data/training/animeface.zip\",\n \"gpus\":1,\n \"augpipe\":\"bg\",\n \"gamma\":10,\n \"cfg\":\"paper256\",\n \"mirror\":1,\n \"snap\":10,\n \"kimg\":2000,\n \"metrics\":\"none\",\n \"outdir\":\"/opt/ml/model\",\n }\n```\n\n本文为了简化工程演示，将“kimg”设为1。\n\n\n### **产品服务**\n\n\n#### **Amazon SageMaker Pipeline**\n\n**Amazon SageMaker Pipelines** 是一种利用 Amazon SageMake 直接 **集成构建机器学习管道的工具**。利用 Pipeline，您可以实现复杂工作流步骤的编排，其中不仅包括机器学习相关的特征工程、训练和部署等步骤，还包括直接集成大规模数据处理的 **EMR Step** 以及可以用来触发其他服务的 **Lambda Step**。配置好的管道可以通过 Amazon EventBridge 触发，或是通过 Lambda 进行调用。同时，您也可以通过配置占位参数更便捷地换元并执行整个管道。如果您使用 SageMaker Studio，还可以动态查看流程的进展以及相关的信息。\n\n\n\n在本文中，我们介绍如何用 Pipeline 串联起 **数据处理**、**训练**、**在线和离线部署** 等步骤来组织整个工作流。结合占位元，该工作流就可以比较灵活的复用。比如，当我们有多个模型以同样的流程提供线上服务，我们就可以将镜像地址作为占位元实现“一键”复用而无需反复配置；又或是我们想尝试不同的资源配置来测试整个流程的性能，那么我们就可以使用相关参数作为占位元。同时，结合 **Amazon EventBridge** 和 **Amazon Lambda**，我们可以实现自动触发机制。比如，每周或是每月会产生一批新的生产数据，而我们需要定时对他们进行批量推理；又或是，每当 S3 特定位置中上传了新文件我们需要重新进行模型评估。在启动整个管道流程后我们可以根据每个组件的 Log Group 在 CloudWatch 上查找相关日志。比如 TrainingStep 相关的日志可以在对应 Training Job 的 Log Group 中查看。\n\nAmazon SageMaker Pipeline：\n\n[https://docs.aws.amazon.com/zh_cn/sagemaker/latest/dg/pipelines-sdk.html]()\n\nAmazon EventBridge：\n\n[https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-what-is.html]()\n\nAmazon Lambda：\n\n[https://docs.aws.amazon.com/lambda/latest/dg/welcome.html]()\n\n#### **SageMaker Processing Job**\n\nSageMaker Processing Job 是一款 **可分布式作业的托管数据处理功能**，您可以利用他来完成例如特征工程、数据验证、模型评估和模型批量推理等任务。您不仅可以选择使用 Sagemaker 预置的处理镜像，也可以选择自带镜像。\n\n\n\n在本文中，我们将用 Processing Job 来完成数据的预处理和模型训练后的批量推理任务。\n\n\n\nSageMaker Processing Job：\n\n[https://docs.aws.amazon.com/sagemaker/latest/dg/processing-job.html]()\n\n#### **Sagemaker Training Job BYOC: Bring Your Own Container(自带镜像实现模型训练)**\n\n通常我们会建议客户以自带算法脚本（Bring Your Own Scripts）的形式在 SageMaker 上进行训练，然而为了最大限度保证生产环境的稳定性和灵活性，在本文中我们将介绍自带镜像在 Amazon SageMaker 上进行训练的方式。在这里我们选择使用 SageMaker 预置的基础镜像来减省环境配置的时间，更多 **预置镜像** 请参考下方链接。\n\nSagemaker Training Job：\n\n[https://docs.aws.amazon.com/sagemaker/latest/dg/how-it-works-training.html]()\n\n预置镜像参考：\n\n[https://github.com/aws/deep-learning-containers/blob/master/available_images.md]()\n\n#### **Amazon ECR**\n\nAmazon Elastic Container Registry (Amazon ECR) 是一款安全、可扩展且可靠的亚马逊云科技 **托管容器映像注册表服务**。Amazon ECR 使用 Amazon IAM 支持具有基于资源的权限的私有存储库。您可以使用首 API 完成推送、拉取和管理 Docker 映像、开放容器倡议 (OCI) 映像和 OCI 兼容的工件。\n\n\n\n在本文中我们将会使用 Amazon ECR 来存储我们的训练镜像。\n\n\n\nAmazon ECR：\n\n[https://docs.aws.amazon.com/AmazonECR/latest/userguide/what-is-ecr.html]()\n\n#### **Amazon SNS**\n\nAmazon Simple Notification Service (Amazon SNS) 是一项托管服务，可提供从发布者到订阅者（也称为生产者和消费者）的消息传递。发布者通过向主题发送消息来与订阅者异步通信，主题是逻辑访问点和通信通道。客户端可以订阅 SNS 主题并使用受支持的终端节点类型接收已发布的消息。\n\n\n\n在本文中 Amazon SNS 将在整个工作流顺利完成之后用邮件告知其主题订阅者最终推理结果在 S3 上的存储位置。如果您还没有创建过主题，请您参考下方链接来 **创建您的 SNS 主题** 并配置订阅邮箱。这样在启动长时间的训练任务时，我们就不必手动去查看任务状态。您还可以利用 SNS 做任务发起通知和故障通知。\n\nAmazon SNS：\n\n[https://docs.aws.amazon.com/sns/latest/dg/welcome.html]()\n\n创建 SNS 主题参考：\n\n[https://docs.aws.amazon.com/sns/latest/dg/sns-create-topic.html]()\n\n\n#### **技术实现**\n\n以下为本文中 Pipeline 的示意图。我们可以比较清晰地看到 Pipeline Steps 和底层功能组件的对应关系。在代码层面，二者的语法入参也基本一致。\n\n\n\n组织 SageMaker Pipeline 更像是搭积木，我们分别定义每一个步骤，最终再将他们串起来。\n\n\n\n第一步我们需要配置 Pipeline 的占位元：\n\n```\nfrom sagemaker.workflow.parameters import ParameterInteger, ParameterString, ParameterFloat\n```\n\n**Process Placeholders**\n\n```\nprocess_input_data_uri = 's3://{}/My-StyleGAN2-Pipeline/animeface/'.format(default_bucket)\nprocess_input_data_path = ParameterString(name=\"ProcessInput\",default_value=process_input_data_uri,)\nprocess_instance_count = ParameterInteger(name=\"ProcessInstanceCount\",default_value=1)\nprocess_instance_type = ParameterString(name=\"ProcessInstancetType\",default_value='ml.m5.xlarge',)\n```\n\n**Train Placeholders**\n\n```\n#image_uri = '357224784104.dkr.ecr.us-west-2.amazonaws.com/blogstylegan2'\nimage_uri = stylegan2_image\noutput_uri = 's3://{}/My-StyleGAN2-Pipeline/Model'.format(default_bucket)\ncheckpoint_uri = 's3://{}/My-StyleGAN2-Pipeline/checkpoints'.format(default_bucket)\n\ntrain_instance_count = ParameterInteger(name=\"TrainInstanceCount\",default_value=1)\ntrain_instance_type = ParameterString(name=\"TrainInstancetType\",default_value='ml.p3.2xlarge',)\ntrain_checkpoint_path = ParameterString(name=\"TrainCheckpointPath\",default_value=checkpoint_uri)\ntrain_output_path = ParameterString(name=\"TrainOutputlPath\",default_value=output_uri)# we write the final model to the same S3 directory as the inferencing source codes\ntrain_image = ParameterString(name=\"TrainImage\",default_value=image_uri,)\n```\n\n\n**Inference Placeholders**\n\n\n```\nsource_code_uri = sagemaker_session.upload_data('stylegan2-ada-pytorch', key_prefix='My-StyleGAN2-Pipeline/Inference')\n#Upload a test image\ninference_img_uri = sagemaker_session.upload_data('test.png', key_prefix='My-StyleGAN2-Pipeline/InferenceImg')\n\ninference_code_path = ParameterString(name=\"InferenceCodePath\",default_value=source_code_uri)\ninference_image_path = ParameterString(name=\"InferenceImgPath\",default_value=inference_img_uri)\ninference_instance_count = ParameterInteger(name=\"InferenceInstanceCount\",default_value=1)\ninference_instance_type = ParameterString(name=\"InferenceInstancetType\",default_value='ml.g4dn.2xlarge',)\n```\n\n第二步我们需要定义数据预处理的 ProcessingStep：\n\n```\nfrom sagemaker.processing import ProcessingInput, ProcessingOutput\nfrom sagemaker.workflow.steps import ProcessingStep\nfrom sagemaker.sklearn.processing import SKLearnProcessor\nframework_version = \"0.23-1\"\n\n# Configure Processor\nsklearn_processor = SKLearnProcessor(\n framework_version=framework_version,\n instance_type=process_instance_type,\n instance_count=process_instance_count,\n role=role,\n)\n\n# Configure ProcessingStep\nstep_process = ProcessingStep(\n name=\"stylegan2Process\",\n processor=sklearn_processor,\n inputs=[\n ProcessingInput(source=process_input_data, destination=\"/opt/ml/processing/input\"), \n ],\n outputs=[\n ProcessingOutput(output_name=\"train\", source=\"/opt/ml/processing/train\")\n ],\n ## Processing Arguments\n job_arguments=['--source', '/opt/ml/processing/input/',\n '--dest','/opt/ml/processing/train/animeface.zip',\n '--width', '256',\n '--height','256',],\n code=\"code_pipeline/dataset_tool.py\",\n)\n```\n\n第三步我们需要定义用作模型训练的 TrainingStep。这边我们可以直接引用上一步数据处理的输出作为本步骤的输入。\n\n\n\n同时我们也可以通过 depends_on 来定义步骤之间的依赖关系。在配置训练任务时，我们推荐您使用 checkpoints，SageMaker 将会自动将 /opt/ml/checkpoints 路径下的文件同步到 S3 指定路径，使您可以及时查看训练的中间结果。在训练结束之后，SageMaker 将会将 /opt/ml/model 下的文件作为最终模型输出，因此我们需要调整代码中模型最后的输出位置。\n\n```\nfrom sagemaker.inputs import TrainingInput\nfrom sagemaker.workflow.steps import TrainingStep\nfrom sagemaker.estimator import Estimator\n\n# Configure training parameters\ndef json_encode_hyperparameters(hyperparameters):\n return {str(k): json.dumps(v) for (k, v) in hyperparameters.items()}\nparams = {\"data\": \"/opt/ml/input/data/train/animeface.zip\",\n \"gpus\": 1,\n \"augpipe\": \"bg\",\n \"gamma\": 10,\n \"cfg\": \"paper256\", \n \"mirror\": 1, \n \"snap\": 10,\n \"metrics\": \"none\",\n \"kimg\": 1,\n \"outdir\": \"/opt/ml/checkpoints\"}\nhyperparameters = json_encode_hyperparameters(params)\n\n# Configure the estimator\nestimator_stylegan2 = Estimator(\n role=role,\n image_uri=train_image,\n train_instance_count=train_instance_count,\n train_instance_type=train_instance_type,\n hyperparameters=hyperparameters,\n disable_profiler=True,\n checkpoint_s3_uri=train_checkpoint_path,\n checkpoint_local_path='/opt/ml/checkpoints',\n output_path= train_output_path,\n)\n\n# Configure Training Step\nstep_train = TrainingStep(\n name=\"stylegan2train\",\n estimator = estimator_stylegan2,\n inputs={\n \"train\": TrainingInput(s3_data=step_process.properties.ProcessingOutputConfig.Outputs[\"train\"].S3Output.S3Uri, \n content_type = 'application/x-image'),\n },\n depends_on = [step_process],\n)\n```\n\n第四步我们需要定义用作批量推理的 ProcessingStep。我们可以按需定义多个数据输入和输出通道。如下所示我们可以分别传入代码，模型和用作推理的样例数据。本文为了工程演示选择实现了原代码中映射 (projector.py) 这个功能，您可以根据自己的需要采样生成 (generate.py) 或是进行风格混合 (style_mixing.py)。\n\n```\n#Initialize the PyTorchProcessor\npytorch_processor = PyTorchProcessor(\n framework_version='1.10.2',\n role=get_execution_role(),\n instance_type=inference_instance_type,\n instance_count=inference_instance_count,\n base_job_name='stylegan2_batch_inference',\n py_version = 'py38'\n)\n\n# Configure ProcessingStep\nstep_process_inference = ProcessingStep(\n name=\"stylegan2Inference\",\n processor=pytorch_processor,\n inputs=[\n # input 1: source code \n ProcessingInput(source=inference_code_path,destination=\"/opt/ml/processing/input\"),\n # input 2: trained model\n ProcessingInput(source=step_train.properties.ModelArtifacts.S3ModelArtifacts,destination=\"/opt/ml/processing/model\"),\n # input 3: test image\n ProcessingInput(source=inference_image_path,destination=\"/opt/ml/processing/data\")\n ],\n outputs=[\n ProcessingOutput(output_name=\"result\", source=\"/opt/ml/processing/output/test\")\n ],\n code=\"code_pipeline/inference.sh\",\n depends_on=[step_train]\n)\n```\n\n最后我们需要定义负责消息分发的 LambdaStep。首先我们需要定义对应的 Lambda Function，该 Function 会在整个工作流顺利结束之后给订阅群组发邮件同时提示他们推理结果在 S3 上的存储位置。\n\n```\n%%writefile lambda_deployer.py\n\n\"\"\"\nThis Lambda function sents Email to SNS Topic end users once the inference is complete and notify them the output directory on S3.\n\"\"\"\n\nimport json\nimport boto3\n\ndef lambda_handler(event, context):\n \"\"\" \"\"\"\n sns_client = boto3.client(\"sns\")\n\n output_s3_dir = event[\"output_s3_dir\"]\n msg = 'The Inference is done! The output has been stored at: '+str(output_s3_dir)\n response = sns_client.publish(\n TopicArn='<Your Topic Arn Here>',\n Message=msg,\n Subject='StyleGAN2 Inference',\n)\n return {\n \"statusCode\": 200,\n \"body\": json.dumps(\"Message Sent!\"),\n }\n```\n\n然后我们需要定义预定的 LambdaStep：\n\n```\n# Define Lambda Step\nfrom sagemaker.lambda_helper import Lambda\nfrom sagemaker.workflow.lambda_step import (\n LambdaStep,\n LambdaOutput,\n LambdaOutputTypeEnum,\n)\n\nfunction_name = \"StyleGAN2_pipeline_callback\"\n\n# Lambda helper class can be used to create the Lambda function\nfunc = Lambda(\n function_name=function_name,\n execution_role_arn=lambda_role,\n script=\"lambda_deployer.py\",\n handler=\"lambda_deployer.lambda_handler\",\n timeout=600,\n memory_size=10240,\n)\n\n# Lambda Step Input\noutput_s3_dir = step_process_inference.properties.ProcessingOutputConfig.Outputs[\"result\"].S3Output.S3Uri\n# Lambda Step Output\noutput_param_1 = LambdaOutput(output_name=\"statusCode\", output_type=LambdaOutputTypeEnum.String)\noutput_param_2 = LambdaOutput(output_name=\"body\", output_type=LambdaOutputTypeEnum.String)\n\nstep_lambda = LambdaStep(\n name=\"stylegan2Notification\",\n lambda_func=func,\n inputs={\n \"output_s3_dir\": output_s3_dir,\n },\n outputs=[output_param_1, output_param_2],\n depends_on=[step_process_inference]\n)\n```\n\n最后我们要将整个工作流组装并执行。这边要提醒大家，当我们的工作流逻辑非常复杂的时候，对整个工作流的代码调试会比较困难，因此建议先单独验证每一个组件再进行整体的串联。\n\n```\nfrom sagemaker.workflow.pipeline import Pipeline\n\npipeline_name = f\"stylegan2pipeline\"\npipeline = Pipeline(\n name=pipeline_name,\n parameters=[process_input_data_path, \n process_instance_count, \n process_instance_type, \n train_instance_count, \n train_instance_type, \n train_checkpoint_path, \n train_output_path,\n train_image,\n inference_image_path,\n inference_code_path,\n inference_instance_count,\n inference_instance_type\n ],\n steps=[step_process, step_train, step_process_inference, step_lambda],\n)\npipeline.upsert(role_arn=role)\nexecution = pipeline.start()\n```\n\nPipeline 开始执行之后，我们可以在 SageMaker Studio 里面可视化看到动态的执行结果和相关日志。\n\n\n\n\n\n\n\n还可以通过换元“一键”重启。\n\n\n\n\n\n### **总结**\n\n随着诸多领域深度学习模型逐渐成熟并达到商用级别，怎么样能够用最小的运维成本来将模型产品化、平台化变成为了首要问题。在这一点上 SageMaker Pipeline 给出了答案：**不仅可以编排复杂的工作流自动化执行，同时还省去了对底层资源的运维，这让相关应用可以快速上线。**\n\n\n\n同时，对于大多数生成类模型，强壮的算力资源是训练和在线推理所必须的，然而本地自建集群往往难以灵活配置机型和数量，而 SageMaker 在这一点上也有非常明显的优势：可以灵活配置算力集群并实现弹性扩展，使得应用架构对业务变化的适应性更高。\n\n### **本篇作者**\n\n\n**李雪晴**\n\nGCR AI/ML 解决方案架构师。\n\n\n\n**粟伟**\n\n亚马逊云科技资深解决方案架构师，专注游戏行业，开源项目爱好者,致力于云原生应用推广、落地。具有15年以上的信息技术行业专业经验，担任过高级软件工程师，系统架构师等职位，在加入亚马逊云科技之前曾就职于 Bea, Oracle, IBM 等公司。\n\n\n","render":"<p><img src=\"https://dev-media.amazoncloud.cn/df699fb883784df98d14562e9b12cd1d_640%20%281%29.gif\" alt=\"640 1.gif\" /></p>\n<h3><a id=\"_2\"></a><strong>场景简介</strong></h3>\n<p>随着深度学习领域的不断发展，很多生成类模型的效果已经达到“准商用”的水平，因此，也有很多客户选择在这类模型上进行调优和封装并作为应用提供给终端用户，从而实现算法能力产品化。</p>\n<p>本文中，我们将为大家展示如何用利用 Amazon SageMaker Pipeline 构建包括 <strong>数据预处理、模型训练、模型推理以及任务状态通知的自动化工作流</strong>。在本文中我们也将为大家展示主要部分的技术实现，您可以参考<strong>完整代码</strong>（<a href=\"\" target=\"_blank\">https://github.com/DearAmber/SageMakerPipelineStyleGAN2</a>）。</p>\n<h3><a id=\"_StyleGAN2_10\"></a><strong>模型介绍: StyleGAN2</strong></h3>\n<p>StyleGAN 是目前最先进的高分辨率图像合成方法之一，它生成的人脸照片一度被认为“逼真到吓人”。在2021年初，英伟达 (NVIDIA) 的研究人员发布并开源了升级版本：StyleGAN2 。该版本重点修复了特征伪影问题，并在进一步提高生成图像质量的同时降低了训练时间和训练集大小。</p>\n<p>自面世以来，StyleGAN2 受到了众多智能视觉爱好者的欢迎，其实践除了人脸生成以外还包括宠物生成、动漫生成、场景生成等等。广泛的应用场景也引起了艺术设计领域从业者的兴趣，他们希望借助此模型来丰富他们现有平台的功能，利用 AI 来赋能内容生成。</p>\n<p>StyleGAN2：</p>\n<p><a href=\"\" target=\"_blank\">https://github.com/NVlabs/stylegan2-ada-pytorch</a></p>\n<h3><a id=\"_Anime_Face_Dataset_24\"></a><strong>数据介绍: Anime Face Dataset</strong></h3>\n<p>Anime Face Dataset 是一个来自于 Kaggle 平台的开源数据集。里面包含63,632张卡通头像图片可用于做各种各样的深度学习实验。</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/0a6b2f32080f420896c8b11713da5e36_image.png\" alt=\"image.png\" /></p>\n<p>Anime Face Dataset：<br />\n<a href=\"\" target=\"_blank\">https://www.kaggle.com/datasets/splcher/animefacedataset</a></p>\n<h3><a id=\"_34\"></a><strong>工程架构参考</strong></h3>\n<p>如下图所示，我们不妨模拟为平台客户封装 AI 增值服务的场景。在该场景中，平台客户的终端用户有可能是原画师或者设计师，他们仅需要在客户的前端网页上上传标注好的图像素材，平台客户则仅需要在后台触发预设好的 SageMaker Pipeline 任务，SageMaker Pipeline 将会自动获取 S3 中的训练数据并开启整个工作流，并按需回传训练的中间结果，最终的模型文件和推理结果。在推理完成之后，SageMaker Pipeline 会通过 Amazon SNS 通知平台终端用户。</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/184ccfad44674c0e8ad3918fec76dc00_image.png\" alt=\"image.png\" /></p>\n<h3><a id=\"_41\"></a><strong>部分实验结果</strong></h3>\n<p>本文的实验框架环境为 Python3.8，PyTorch1.8.1+cu111。同时选用了单个 ml.p3.2xlarge 作为训练计算资源。</p>\n<p>下图为完整训练后部分生成样本，仅供参考。</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/8701c30ba248448fbc59325cd460863d_image.png\" alt=\"image.png\" /></p>\n<p>完整的训练将需要持续72小时左右，请参考以下超参：</p>\n<pre><code class=\"lang-\">hyperparameters = {&quot;data&quot;:&quot;/opt/ml/input/data/training/animeface.zip&quot;,\n &quot;gpus&quot;:1,\n &quot;augpipe&quot;:&quot;bg&quot;,\n &quot;gamma&quot;:10,\n &quot;cfg&quot;:&quot;paper256&quot;,\n &quot;mirror&quot;:1,\n &quot;snap&quot;:10,\n &quot;kimg&quot;:2000,\n &quot;metrics&quot;:&quot;none&quot;,\n &quot;outdir&quot;:&quot;/opt/ml/model&quot;,\n }\n</code></pre>\n<p>本文为了简化工程演示，将“kimg”设为1。</p>\n<h3><a id=\"_70\"></a><strong>产品服务</strong></h3>\n<h4><a id=\"Amazon_SageMaker_Pipeline_73\"></a><strong>Amazon SageMaker Pipeline</strong></h4>\n<p><strong>Amazon SageMaker Pipelines</strong> 是一种利用 Amazon SageMake 直接 <strong>集成构建机器学习管道的工具</strong>。利用 Pipeline，您可以实现复杂工作流步骤的编排，其中不仅包括机器学习相关的特征工程、训练和部署等步骤，还包括直接集成大规模数据处理的 <strong>EMR Step</strong> 以及可以用来触发其他服务的 <strong>Lambda Step</strong>。配置好的管道可以通过 Amazon EventBridge 触发，或是通过 Lambda 进行调用。同时，您也可以通过配置占位参数更便捷地换元并执行整个管道。如果您使用 SageMaker Studio，还可以动态查看流程的进展以及相关的信息。</p>\n<p>在本文中，我们介绍如何用 Pipeline 串联起 <strong>数据处理</strong>、<strong>训练</strong>、<strong>在线和离线部署</strong> 等步骤来组织整个工作流。结合占位元，该工作流就可以比较灵活的复用。比如，当我们有多个模型以同样的流程提供线上服务，我们就可以将镜像地址作为占位元实现“一键”复用而无需反复配置；又或是我们想尝试不同的资源配置来测试整个流程的性能，那么我们就可以使用相关参数作为占位元。同时，结合 <strong>Amazon EventBridge</strong> 和 <strong>Amazon Lambda</strong>，我们可以实现自动触发机制。比如，每周或是每月会产生一批新的生产数据，而我们需要定时对他们进行批量推理；又或是，每当 S3 特定位置中上传了新文件我们需要重新进行模型评估。在启动整个管道流程后我们可以根据每个组件的 Log Group 在 CloudWatch 上查找相关日志。比如 TrainingStep 相关的日志可以在对应 Training Job 的 Log Group 中查看。</p>\n<p>Amazon SageMaker Pipeline：</p>\n<p><a href=\"\" target=\"_blank\">https://docs.aws.amazon.com/zh_cn/sagemaker/latest/dg/pipelines-sdk.html</a></p>\n<p>Amazon EventBridge：</p>\n<p><a href=\"\" target=\"_blank\">https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-what-is.html</a></p>\n<p>Amazon Lambda：</p>\n<p><a href=\"\" target=\"_blank\">https://docs.aws.amazon.com/lambda/latest/dg/welcome.html</a></p>\n<h4><a id=\"SageMaker_Processing_Job_93\"></a><strong>SageMaker Processing Job</strong></h4>\n<p>SageMaker Processing Job 是一款 <strong>可分布式作业的托管数据处理功能</strong>，您可以利用他来完成例如特征工程、数据验证、模型评估和模型批量推理等任务。您不仅可以选择使用 Sagemaker 预置的处理镜像，也可以选择自带镜像。</p>\n<p>在本文中，我们将用 Processing Job 来完成数据的预处理和模型训练后的批量推理任务。</p>\n<p>SageMaker Processing Job：</p>\n<p><a href=\"\" target=\"_blank\">https://docs.aws.amazon.com/sagemaker/latest/dg/processing-job.html</a></p>\n<h4><a id=\"Sagemaker_Training_Job_BYOC_Bring_Your_Own_Container_107\"></a><strong>Sagemaker Training Job BYOC: Bring Your Own Container(自带镜像实现模型训练)</strong></h4>\n<p>通常我们会建议客户以自带算法脚本（Bring Your Own Scripts）的形式在 SageMaker 上进行训练，然而为了最大限度保证生产环境的稳定性和灵活性，在本文中我们将介绍自带镜像在 Amazon SageMaker 上进行训练的方式。在这里我们选择使用 SageMaker 预置的基础镜像来减省环境配置的时间，更多 <strong>预置镜像</strong> 请参考下方链接。</p>\n<p>Sagemaker Training Job：</p>\n<p><a href=\"\" target=\"_blank\">https://docs.aws.amazon.com/sagemaker/latest/dg/how-it-works-training.html</a></p>\n<p>预置镜像参考：</p>\n<p><a href=\"\" target=\"_blank\">https://github.com/aws/deep-learning-containers/blob/master/available_images.md</a></p>\n<h4><a id=\"Amazon_ECR_119\"></a><strong>Amazon ECR</strong></h4>\n<p>Amazon Elastic Container Registry (Amazon ECR) 是一款安全、可扩展且可靠的亚马逊云科技 <strong>托管容器映像注册表服务</strong>。Amazon ECR 使用 Amazon IAM 支持具有基于资源的权限的私有存储库。您可以使用首 API 完成推送、拉取和管理 Docker 映像、开放容器倡议 (OCI) 映像和 OCI 兼容的工件。</p>\n<p>在本文中我们将会使用 Amazon ECR 来存储我们的训练镜像。</p>\n<p>Amazon ECR：</p>\n<p><a href=\"\" target=\"_blank\">https://docs.aws.amazon.com/AmazonECR/latest/userguide/what-is-ecr.html</a></p>\n<h4><a id=\"Amazon_SNS_133\"></a><strong>Amazon SNS</strong></h4>\n<p>Amazon Simple Notification Service (Amazon SNS) 是一项托管服务，可提供从发布者到订阅者（也称为生产者和消费者）的消息传递。发布者通过向主题发送消息来与订阅者异步通信，主题是逻辑访问点和通信通道。客户端可以订阅 SNS 主题并使用受支持的终端节点类型接收已发布的消息。</p>\n<p>在本文中 Amazon SNS 将在整个工作流顺利完成之后用邮件告知其主题订阅者最终推理结果在 S3 上的存储位置。如果您还没有创建过主题，请您参考下方链接来 <strong>创建您的 SNS 主题</strong> 并配置订阅邮箱。这样在启动长时间的训练任务时，我们就不必手动去查看任务状态。您还可以利用 SNS 做任务发起通知和故障通知。</p>\n<p>Amazon SNS：</p>\n<p><a href=\"\" target=\"_blank\">https://docs.aws.amazon.com/sns/latest/dg/welcome.html</a></p>\n<p>创建 SNS 主题参考：</p>\n<p><a href=\"\" target=\"_blank\">https://docs.aws.amazon.com/sns/latest/dg/sns-create-topic.html</a></p>\n<h4><a id=\"_150\"></a><strong>技术实现</strong></h4>\n<p>以下为本文中 Pipeline 的示意图。我们可以比较清晰地看到 Pipeline Steps 和底层功能组件的对应关系。在代码层面，二者的语法入参也基本一致。</p>\n<p>组织 SageMaker Pipeline 更像是搭积木，我们分别定义每一个步骤，最终再将他们串起来。</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/d1c99e83a5854f94be09dafe5c441aa2_image.png\" alt=\"image.png\" /></p>\n<p>第一步我们需要配置 Pipeline 的占位元：</p>\n<pre><code class=\"lang-\">from sagemaker.workflow.parameters import ParameterInteger, ParameterString, ParameterFloat\n</code></pre>\n<p><strong>Process Placeholders</strong></p>\n<pre><code class=\"lang-\">process_input_data_uri = 's3://{}/My-StyleGAN2-Pipeline/animeface/'.format(default_bucket)\nprocess_input_data_path = ParameterString(name=&quot;ProcessInput&quot;,default_value=process_input_data_uri,)\nprocess_instance_count = ParameterInteger(name=&quot;ProcessInstanceCount&quot;,default_value=1)\nprocess_instance_type = ParameterString(name=&quot;ProcessInstancetType&quot;,default_value='ml.m5.xlarge',)\n</code></pre>\n<p><strong>Train Placeholders</strong></p>\n<pre><code class=\"lang-\">#image_uri = '357224784104.dkr.ecr.us-west-2.amazonaws.com/blogstylegan2'\nimage_uri = stylegan2_image\noutput_uri = 's3://{}/My-StyleGAN2-Pipeline/Model'.format(default_bucket)\ncheckpoint_uri = 's3://{}/My-StyleGAN2-Pipeline/checkpoints'.format(default_bucket)\n\ntrain_instance_count = ParameterInteger(name=&quot;TrainInstanceCount&quot;,default_value=1)\ntrain_instance_type = ParameterString(name=&quot;TrainInstancetType&quot;,default_value='ml.p3.2xlarge',)\ntrain_checkpoint_path = ParameterString(name=&quot;TrainCheckpointPath&quot;,default_value=checkpoint_uri)\ntrain_output_path = ParameterString(name=&quot;TrainOutputlPath&quot;,default_value=output_uri)# we write the final model to the same S3 directory as the inferencing source codes\ntrain_image = ParameterString(name=&quot;TrainImage&quot;,default_value=image_uri,)\n</code></pre>\n<p><strong>Inference Placeholders</strong></p>\n<pre><code class=\"lang-\">source_code_uri = sagemaker_session.upload_data('stylegan2-ada-pytorch', key_prefix='My-StyleGAN2-Pipeline/Inference')\n#Upload a test image\ninference_img_uri = sagemaker_session.upload_data('test.png', key_prefix='My-StyleGAN2-Pipeline/InferenceImg')\n\ninference_code_path = ParameterString(name=&quot;InferenceCodePath&quot;,default_value=source_code_uri)\ninference_image_path = ParameterString(name=&quot;InferenceImgPath&quot;,default_value=inference_img_uri)\ninference_instance_count = ParameterInteger(name=&quot;InferenceInstanceCount&quot;,default_value=1)\ninference_instance_type = ParameterString(name=&quot;InferenceInstancetType&quot;,default_value='ml.g4dn.2xlarge',)\n</code></pre>\n<p>第二步我们需要定义数据预处理的 ProcessingStep：</p>\n<pre><code class=\"lang-\">from sagemaker.processing import ProcessingInput, ProcessingOutput\nfrom sagemaker.workflow.steps import ProcessingStep\nfrom sagemaker.sklearn.processing import SKLearnProcessor\nframework_version = &quot;0.23-1&quot;\n\n# Configure Processor\nsklearn_processor = SKLearnProcessor(\n framework_version=framework_version,\n instance_type=process_instance_type,\n instance_count=process_instance_count,\n role=role,\n)\n\n# Configure ProcessingStep\nstep_process = ProcessingStep(\n name=&quot;stylegan2Process&quot;,\n processor=sklearn_processor,\n inputs=[\n ProcessingInput(source=process_input_data, destination=&quot;/opt/ml/processing/input&quot;), \n ],\n outputs=[\n ProcessingOutput(output_name=&quot;train&quot;, source=&quot;/opt/ml/processing/train&quot;)\n ],\n ## Processing Arguments\n job_arguments=['--source', '/opt/ml/processing/input/',\n '--dest','/opt/ml/processing/train/animeface.zip',\n '--width', '256',\n '--height','256',],\n code=&quot;code_pipeline/dataset_tool.py&quot;,\n)\n</code></pre>\n<p>第三步我们需要定义用作模型训练的 TrainingStep。这边我们可以直接引用上一步数据处理的输出作为本步骤的输入。</p>\n<p>同时我们也可以通过 depends_on 来定义步骤之间的依赖关系。在配置训练任务时，我们推荐您使用 checkpoints，SageMaker 将会自动将 /opt/ml/checkpoints 路径下的文件同步到 S3 指定路径，使您可以及时查看训练的中间结果。在训练结束之后，SageMaker 将会将 /opt/ml/model 下的文件作为最终模型输出，因此我们需要调整代码中模型最后的输出位置。</p>\n<pre><code class=\"lang-\">from sagemaker.inputs import TrainingInput\nfrom sagemaker.workflow.steps import TrainingStep\nfrom sagemaker.estimator import Estimator\n\n# Configure training parameters\ndef json_encode_hyperparameters(hyperparameters):\n return {str(k): json.dumps(v) for (k, v) in hyperparameters.items()}\nparams = {&quot;data&quot;: &quot;/opt/ml/input/data/train/animeface.zip&quot;,\n &quot;gpus&quot;: 1,\n &quot;augpipe&quot;: &quot;bg&quot;,\n &quot;gamma&quot;: 10,\n &quot;cfg&quot;: &quot;paper256&quot;, \n &quot;mirror&quot;: 1, \n &quot;snap&quot;: 10,\n &quot;metrics&quot;: &quot;none&quot;,\n &quot;kimg&quot;: 1,\n &quot;outdir&quot;: &quot;/opt/ml/checkpoints&quot;}\nhyperparameters = json_encode_hyperparameters(params)\n\n# Configure the estimator\nestimator_stylegan2 = Estimator(\n role=role,\n image_uri=train_image,\n train_instance_count=train_instance_count,\n train_instance_type=train_instance_type,\n hyperparameters=hyperparameters,\n disable_profiler=True,\n checkpoint_s3_uri=train_checkpoint_path,\n checkpoint_local_path='/opt/ml/checkpoints',\n output_path= train_output_path,\n)\n\n# Configure Training Step\nstep_train = TrainingStep(\n name=&quot;stylegan2train&quot;,\n estimator = estimator_stylegan2,\n inputs={\n &quot;train&quot;: TrainingInput(s3_data=step_process.properties.ProcessingOutputConfig.Outputs[&quot;train&quot;].S3Output.S3Uri, \n content_type = 'application/x-image'),\n },\n depends_on = [step_process],\n)\n</code></pre>\n<p>第四步我们需要定义用作批量推理的 ProcessingStep。我们可以按需定义多个数据输入和输出通道。如下所示我们可以分别传入代码，模型和用作推理的样例数据。本文为了工程演示选择实现了原代码中映射 (projector.py) 这个功能，您可以根据自己的需要采样生成 (generate.py) 或是进行风格混合 (style_mixing.py)。</p>\n<pre><code class=\"lang-\">#Initialize the PyTorchProcessor\npytorch_processor = PyTorchProcessor(\n framework_version='1.10.2',\n role=get_execution_role(),\n instance_type=inference_instance_type,\n instance_count=inference_instance_count,\n base_job_name='stylegan2_batch_inference',\n py_version = 'py38'\n)\n\n# Configure ProcessingStep\nstep_process_inference = ProcessingStep(\n name=&quot;stylegan2Inference&quot;,\n processor=pytorch_processor,\n inputs=[\n # input 1: source code \n ProcessingInput(source=inference_code_path,destination=&quot;/opt/ml/processing/input&quot;),\n # input 2: trained model\n ProcessingInput(source=step_train.properties.ModelArtifacts.S3ModelArtifacts,destination=&quot;/opt/ml/processing/model&quot;),\n # input 3: test image\n ProcessingInput(source=inference_image_path,destination=&quot;/opt/ml/processing/data&quot;)\n ],\n outputs=[\n ProcessingOutput(output_name=&quot;result&quot;, source=&quot;/opt/ml/processing/output/test&quot;)\n ],\n code=&quot;code_pipeline/inference.sh&quot;,\n depends_on=[step_train]\n)\n</code></pre>\n<p>最后我们需要定义负责消息分发的 LambdaStep。首先我们需要定义对应的 Lambda Function，该 Function 会在整个工作流顺利结束之后给订阅群组发邮件同时提示他们推理结果在 S3 上的存储位置。</p>\n<pre><code class=\"lang-\">%%writefile lambda_deployer.py\n\n&quot;&quot;&quot;\nThis Lambda function sents Email to SNS Topic end users once the inference is complete and notify them the output directory on S3.\n&quot;&quot;&quot;\n\nimport json\nimport boto3\n\ndef lambda_handler(event, context):\n &quot;&quot;&quot; &quot;&quot;&quot;\n sns_client = boto3.client(&quot;sns&quot;)\n\n output_s3_dir = event[&quot;output_s3_dir&quot;]\n msg = 'The Inference is done! The output has been stored at: '+str(output_s3_dir)\n response = sns_client.publish(\n TopicArn='&lt;Your Topic Arn Here&gt;',\n Message=msg,\n Subject='StyleGAN2 Inference',\n)\n return {\n &quot;statusCode&quot;: 200,\n &quot;body&quot;: json.dumps(&quot;Message Sent!&quot;),\n }\n</code></pre>\n<p>然后我们需要定义预定的 LambdaStep：</p>\n<pre><code class=\"lang-\"># Define Lambda Step\nfrom sagemaker.lambda_helper import Lambda\nfrom sagemaker.workflow.lambda_step import (\n LambdaStep,\n LambdaOutput,\n LambdaOutputTypeEnum,\n)\n\nfunction_name = &quot;StyleGAN2_pipeline_callback&quot;\n\n# Lambda helper class can be used to create the Lambda function\nfunc = Lambda(\n function_name=function_name,\n execution_role_arn=lambda_role,\n script=&quot;lambda_deployer.py&quot;,\n handler=&quot;lambda_deployer.lambda_handler&quot;,\n timeout=600,\n memory_size=10240,\n)\n\n# Lambda Step Input\noutput_s3_dir = step_process_inference.properties.ProcessingOutputConfig.Outputs[&quot;result&quot;].S3Output.S3Uri\n# Lambda Step Output\noutput_param_1 = LambdaOutput(output_name=&quot;statusCode&quot;, output_type=LambdaOutputTypeEnum.String)\noutput_param_2 = LambdaOutput(output_name=&quot;body&quot;, output_type=LambdaOutputTypeEnum.String)\n\nstep_lambda = LambdaStep(\n name=&quot;stylegan2Notification&quot;,\n lambda_func=func,\n inputs={\n &quot;output_s3_dir&quot;: output_s3_dir,\n },\n outputs=[output_param_1, output_param_2],\n depends_on=[step_process_inference]\n)\n</code></pre>\n<p>最后我们要将整个工作流组装并执行。这边要提醒大家，当我们的工作流逻辑非常复杂的时候，对整个工作流的代码调试会比较困难，因此建议先单独验证每一个组件再进行整体的串联。</p>\n<pre><code class=\"lang-\">from sagemaker.workflow.pipeline import Pipeline\n\npipeline_name = f&quot;stylegan2pipeline&quot;\npipeline = Pipeline(\n name=pipeline_name,\n parameters=[process_input_data_path, \n process_instance_count, \n process_instance_type, \n train_instance_count, \n train_instance_type, \n train_checkpoint_path, \n train_output_path,\n train_image,\n inference_image_path,\n inference_code_path,\n inference_instance_count,\n inference_instance_type\n ],\n steps=[step_process, step_train, step_process_inference, step_lambda],\n)\npipeline.upsert(role_arn=role)\nexecution = pipeline.start()\n</code></pre>\n<p>Pipeline 开始执行之后，我们可以在 SageMaker Studio 里面可视化看到动态的执行结果和相关日志。</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/57f9556f21024512939a5d0e102fc067_image.png\" alt=\"image.png\" /></p>\n<p><img src=\"https://dev-media.amazoncloud.cn/7446f5b3ab4a4014beb5b6a401df3d5f_image.png\" alt=\"image.png\" /></p>\n<p>还可以通过换元“一键”重启。</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/fd406e24c4084a2e8f9b2a52e8972afe_image.png\" alt=\"image.png\" /></p>\n<h3><a id=\"_434\"></a><strong>总结</strong></h3>\n<p>随着诸多领域深度学习模型逐渐成熟并达到商用级别，怎么样能够用最小的运维成本来将模型产品化、平台化变成为了首要问题。在这一点上 SageMaker Pipeline 给出了答案：<strong>不仅可以编排复杂的工作流自动化执行，同时还省去了对底层资源的运维，这让相关应用可以快速上线。</strong></p>\n<p>同时，对于大多数生成类模型，强壮的算力资源是训练和在线推理所必须的，然而本地自建集群往往难以灵活配置机型和数量，而 SageMaker 在这一点上也有非常明显的优势：可以灵活配置算力集群并实现弹性扩展，使得应用架构对业务变化的适应性更高。</p>\n<h3><a id=\"_442\"></a><strong>本篇作者</strong></h3>\n<p><img src=\"https://dev-media.amazoncloud.cn/97d185fadae54b48a9a719369d91ff84_image.png\" alt=\"image.png\" /></p>\n<p><strong>李雪晴</strong></p>\n<p>GCR AI/ML 解决方案架构师。</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/82f21d3c4af24ebe92f0a1d6ef36c670_image.png\" alt=\"image.png\" /></p>\n<p><strong>粟伟</strong></p>\n<p>亚马逊云科技资深解决方案架构师，专注游戏行业，开源项目爱好者,致力于云原生应用推广、落地。具有15年以上的信息技术行业专业经验，担任过高级软件工程师，系统架构师等职位，在加入亚马逊云科技之前曾就职于 Bea, Oracle, IBM 等公司。</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/6beeffc802e242eda9063a3c225aa117_640.gif\" alt=\"640.gif\" /></p>\n"}