定义管道

1. 按照 启动亚马逊 SageMaker Studio 中的说明打开 Studio 管理控制台。

2. 在左侧导航窗格中，选择 Pipelines。

3. 选择创建。

4. 选择空白。

5. 在左侧边栏中选择处理数据，然后将其拖到画布上。

6. 在画布中，选择添加的处理数据步骤。

7. 要添加输入数据集，请在右侧边栏的数据（输入）下选择添加，然后选择一个数据集。

8. 要添加保存输出数据集的位置，请在右侧边栏的数据（输出）下选择添加，然后导航至目的地。

9. 填写右侧边栏中的其余字段。有关这些选项卡中字段的信息，请参阅 sagemaker.workflow.steps。 ProcessingStep。

1. 在左侧边栏中选择训练模型，然后将其拖到画布上。

2. 在画布中选择添加的训练模型步骤。

3. 要添加输入数据集，请在右侧边栏的数据（输入）下选择添加，然后选择一个数据集。

4. 要选择保存模型工件的位置，请在位置 (S3URI) 字段URI中输入 Amazon S3，或者选择浏览 S3 导航到目标位置。

5. 填写右侧边栏中的其余字段。有关这些选项卡中字段的信息，请参阅 sagemaker.workflow.steps。 TrainingStep。

6. 单击并拖动光标，从上一节添加的处理数据步骤到训练模型步骤，创建连接两个步骤的边缘。

1. 在左侧边栏中选择注册模型，然后将其拖到画布上。

2. 在画布中，选择添加的注册模型步骤。

3. 要选择要注册的模型，请在模型（输入）下选择添加。

4. 选择创建模型组，将模型添加到新的模型组中。

5. 填写右侧边栏中的其余字段。有关这些选项卡中字段的信息，请参阅 sagemaker.workflow.step_collections。 RegisterModel。

6. 单击并拖动光标，从上一节添加的训练模型步骤到注册模型步骤，创建连接两个步骤的边缘。

1. 在左侧边栏中选择部署模型（端点），然后将其拖到画布上。

2. 在画布中，选择添加的部署模型（端点）步骤。

3. 要选择要部署的模型，请在模型（输入）下选择添加。

4. 选择创建端点单选按钮创建新端点。

5. 为端点输入姓名和描述。

6. 单击并拖动光标，从上一节添加的注册模型步骤到部署模型（端点）步骤，创建连接两个步骤的边缘。

7. 填写右侧边栏中的其余字段。

下载数据集

1. 将数据集下载到您账户的默认 Amazon S3 存储桶中。

   !mkdir -p data
   local_path = "data/abalone-dataset.csv"
   
   s3 = boto3.resource("s3")
   s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file(
       "dataset/abalone-dataset.csv",
       local_path
   )
   
   base_uri = f"s3://{default_bucket}/abalone"
   input_data_uri = sagemaker.s3.S3Uploader.upload(
       local_path=local_path, 
       desired_s3_uri=base_uri,
   )
   print(input_data_uri)

2. 创建模型后，下载第二个数据集进行批量转换。

   local_path = "data/abalone-dataset-batch.csv"
   
   s3 = boto3.resource("s3")
   s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file(
       "dataset/abalone-dataset-batch",
       local_path
   )
   
   base_uri = f"s3://{default_bucket}/abalone"
   batch_data_uri = sagemaker.s3.S3Uploader.upload(
       local_path=local_path, 
       desired_s3_uri=base_uri,
   )
   print(batch_data_uri)

创建处理步骤

1. 为处理脚本创建目录。

   !mkdir -p abalone

2. 在 /abalone 目录中创建一个包含以下内容的名为 preprocessing.py 的文件。该预处理脚本将被传入处理步骤，以便在输入数据上运行。然后，训练步骤使用预处理的训练功能和标签来训练模型。评估步骤使用训练过的模型和预处理过的测试功能和标签对模型进行评估。该脚本使用 scikit-learn 执行以下操作：

   • 填入缺失的 sex 分类数据并对其进行编码，使其适合训练。

   • 缩放和标准化除 rings 和 sex 之外的所有数值字段。

   • 将数据拆分为训练、测试和验证数据集。

   %%writefile abalone/preprocessing.py
   import argparse
   import os
   import requests
   import tempfile
   import numpy as np
   import pandas as pd
   
   
   from sklearn.compose import ColumnTransformer
   from sklearn.impute import SimpleImputer
   from sklearn.pipeline import Pipeline
   from sklearn.preprocessing import StandardScaler, OneHotEncoder
   
   
   # Because this is a headerless CSV file, specify the column names here.
   feature_columns_names = [
       "sex",
       "length",
       "diameter",
       "height",
       "whole_weight",
       "shucked_weight",
       "viscera_weight",
       "shell_weight",
   ]
   label_column = "rings"
   
   feature_columns_dtype = {
       "sex": str,
       "length": np.float64,
       "diameter": np.float64,
       "height": np.float64,
       "whole_weight": np.float64,
       "shucked_weight": np.float64,
       "viscera_weight": np.float64,
       "shell_weight": np.float64
   }
   label_column_dtype = {"rings": np.float64}
   
   
   def merge_two_dicts(x, y):
       z = x.copy()
       z.update(y)
       return z
   
   
   if __name__ == "__main__":
       base_dir = "/opt/ml/processing"
   
       df = pd.read_csv(
           f"{base_dir}/input/abalone-dataset.csv",
           header=None, 
           names=feature_columns_names + [label_column],
           dtype=merge_two_dicts(feature_columns_dtype, label_column_dtype)
       )
       numeric_features = list(feature_columns_names)
       numeric_features.remove("sex")
       numeric_transformer = Pipeline(
           steps=[
               ("imputer", SimpleImputer(strategy="median")),
               ("scaler", StandardScaler())
           ]
       )
   
       categorical_features = ["sex"]
       categorical_transformer = Pipeline(
           steps=[
               ("imputer", SimpleImputer(strategy="constant", fill_value="missing")),
               ("onehot", OneHotEncoder(handle_unknown="ignore"))
           ]
       )
   
       preprocess = ColumnTransformer(
           transformers=[
               ("num", numeric_transformer, numeric_features),
               ("cat", categorical_transformer, categorical_features)
           ]
       )
       
       y = df.pop("rings")
       X_pre = preprocess.fit_transform(df)
       y_pre = y.to_numpy().reshape(len(y), 1)
       
       X = np.concatenate((y_pre, X_pre), axis=1)
       
       np.random.shuffle(X)
       train, validation, test = np.split(X, [int(.7*len(X)), int(.85*len(X))])
   
       
       pd.DataFrame(train).to_csv(f"{base_dir}/train/train.csv", header=False, index=False)
       pd.DataFrame(validation).to_csv(f"{base_dir}/validation/validation.csv", header=False, index=False)
       pd.DataFrame(test).to_csv(f"{base_dir}/test/test.csv", header=False, index=False)

3. 创建要传递到处理步骤的 SKLearnProcessor 的实例。

   from sagemaker.sklearn.processing import SKLearnProcessor
   
   
   framework_version = "0.23-1"
   
   sklearn_processor = SKLearnProcessor(
       framework_version=framework_version,
       instance_type="ml.m5.xlarge",
       instance_count=processing_instance_count,
       base_job_name="sklearn-abalone-process",
       sagemaker_session=pipeline_session,
       role=role,
   )

4. 创建处理步骤。此步骤采用 SKLearnProcessor、输入和输出通道以及您创建的 preprocessing.py 脚本。这与 SageMaker AI Python 中处理器实例的run方法非常相似SDK。传入 ProcessingStep 的 input_data 参数是步骤本身的输入数据。处理器实例运行时会使用这些输入数据。

   请注意在处理作业的输出配置中指定的 "train、"validation 和 "test" 命名通道。这样的步骤 Properties 可以在后续步骤中使用，并在运行时解析为运行时值。

   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="abalone/preprocessing.py",
   ) 
   
   step_process = ProcessingStep(
       name="AbaloneProcess",
       step_args=processor_args
   )

定义训练步骤

1. 指定要保存训练模型的模型路径。

   model_path = f"s3://{default_bucket}/AbaloneTrain"

2. 为XGBoost算法和输入数据集配置估计器。训练实例类型传递到估算器中。一个典型的训练脚本：

   • 从输入通道加载数据

   • 配置超参数训练

   • 训练模型

   • 将模型保存到 model_dir，以便日后托管

   SageMaker AI 在训练作业结束model.tar.gz时以 a 的形式将模型上传到 Amazon S3。

   from sagemaker.estimator import Estimator
   
   
   image_uri = sagemaker.image_uris.retrieve(
       framework="xgboost",
       region=region,
       version="1.0-1",
       py_version="py3",
       instance_type="ml.m5.xlarge"
   )
   xgb_train = Estimator(
       image_uri=image_uri,
       instance_type="ml.m5.xlarge",
       instance_count=1,
       output_path=model_path,
       sagemaker_session=pipeline_session,
       role=role,
   )
   xgb_train.set_hyperparameters(
       objective="reg:linear",
       num_round=50,
       max_depth=5,
       eta=0.2,
       gamma=4,
       min_child_weight=6,
       subsample=0.7,
       silent=0
   )

3. 使用估计器实例和 ProcessingStep 的属性创建一个 TrainingStep。将 "train" 的 S3Uri 和 "validation" 输出通道传递给 TrainingStep。 

   from sagemaker.inputs import TrainingInput
   from sagemaker.workflow.steps import TrainingStep
   
   
   train_args = xgb_train.fit(
       inputs={
           "train": TrainingInput(
               s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                   "train"
               ].S3Output.S3Uri,
               content_type="text/csv"
           ),
           "validation": TrainingInput(
               s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                   "validation"
               ].S3Output.S3Uri,
               content_type="text/csv"
           )
       },
   )
   
   step_train = TrainingStep(
       name="AbaloneTrain",
       step_args = train_args
   )

定义模型评估的处理步骤

1. 在名为 evaluation.py 的 /abalone 目录中创建一个文件。此脚本在处理步骤中用于执行模型评估。它以经过训练的模型和测试数据集作为输入，然后生成包含分类评估指标的JSON文件。

   %%writefile abalone/evaluation.py
   import json
   import pathlib
   import pickle
   import tarfile
   import joblib
   import numpy as np
   import pandas as pd
   import xgboost
   
   
   from sklearn.metrics import mean_squared_error
   
   
   if __name__ == "__main__":
       model_path = f"/opt/ml/processing/model/model.tar.gz"
       with tarfile.open(model_path) as tar:
           tar.extractall(path=".")
       
       model = pickle.load(open("xgboost-model", "rb"))
   
       test_path = "/opt/ml/processing/test/test.csv"
       df = pd.read_csv(test_path, header=None)
       
       y_test = df.iloc[:, 0].to_numpy()
       df.drop(df.columns[0], axis=1, inplace=True)
       
       X_test = xgboost.DMatrix(df.values)
       
       predictions = model.predict(X_test)
   
       mse = mean_squared_error(y_test, predictions)
       std = np.std(y_test - predictions)
       report_dict = {
           "regression_metrics": {
               "mse": {
                   "value": mse,
                   "standard_deviation": std
               },
           },
       }
   
       output_dir = "/opt/ml/processing/evaluation"
       pathlib.Path(output_dir).mkdir(parents=True, exist_ok=True)
       
       evaluation_path = f"{output_dir}/evaluation.json"
       with open(evaluation_path, "w") as f:
           f.write(json.dumps(report_dict))

2. 创建 ScriptProcessor 的实例，用于创建 ProcessingStep。

   from sagemaker.processing import ScriptProcessor
   
   
   script_eval = ScriptProcessor(
       image_uri=image_uri,
       command=["python3"],
       instance_type="ml.m5.xlarge",
       instance_count=1,
       base_job_name="script-abalone-eval",
       sagemaker_session=pipeline_session,
       role=role,
   )

3. 创建一个使用处理实例、输入和输出通道以及 evaluation.py 脚本的处理器的 ProcessingStep。传入：

   • step_train 训练步骤中的 S3ModelArtifacts 属性

   • step_process 处理步骤的 "test" 输出通道的 S3Uri

   这与 SageMaker AI Python 中处理器实例的run方法非常相似SDK。 

   from sagemaker.workflow.properties import PropertyFile
   
   
   evaluation_report = PropertyFile(
       name="EvaluationReport",
       output_name="evaluation",
       path="evaluation.json"
   )
   
   eval_args = script_eval.run(
           inputs=[
           ProcessingInput(
               source=step_train.properties.ModelArtifacts.S3ModelArtifacts,
               destination="/opt/ml/processing/model"
           ),
           ProcessingInput(
               source=step_process.properties.ProcessingOutputConfig.Outputs[
                   "test"
               ].S3Output.S3Uri,
               destination="/opt/ml/processing/test"
           )
       ],
       outputs=[
           ProcessingOutput(output_name="evaluation", source="/opt/ml/processing/evaluation"),
       ],
       code="abalone/evaluation.py",
   )
   
   step_eval = ProcessingStep(
       name="AbaloneEval",
       step_args=eval_args,
       property_files=[evaluation_report],
   )

CreateModelStep 为批量转换定义一个

1. 创建 A SageMaker I 模型。从 step_train 训练步骤传入 S3ModelArtifacts 属性。

   from sagemaker.model import Model
   
   
   model = Model(
       image_uri=image_uri,
       model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
       sagemaker_session=pipeline_session,
       role=role,
   )

2. 为 SageMaker AI 模型定义模型输入。

   from sagemaker.inputs import CreateModelInput
   
   
   inputs = CreateModelInput(
       instance_type="ml.m5.large",
       accelerator_type="ml.eia1.medium",
   )

3. CreateModelStep使用您定义的CreateModelInput和 SageMaker AI 模型实例创建您的。

   from sagemaker.workflow.steps import CreateModelStep
   
   
   step_create_model = CreateModelStep(
       name="AbaloneCreateModel",
       model=model,
       inputs=inputs,
   )

要定义 TransformStep 要执行批量转换

1. 使用适当的计算实例类型、实例数量和所需的输出 Amazon S3 存储桶创建转换器实例URI。从 step_create_model CreateModel 步骤传入 ModelName 属性。

   from sagemaker.transformer import Transformer
   
   
   transformer = Transformer(
       model_name=step_create_model.properties.ModelName,
       instance_type="ml.m5.xlarge",
       instance_count=1,
       output_path=f"s3://{default_bucket}/AbaloneTransform"
   )

2. 使用您定义的转换器实例和 batch_data 管道参数创建 TransformStep。

   from sagemaker.inputs import TransformInput
   from sagemaker.workflow.steps import TransformStep
   
   
   step_transform = TransformStep(
       name="AbaloneTransform",
       transformer=transformer,
       inputs=TransformInput(data=batch_data)
   )

定义创建模型包的 RegisterModel 步骤

• 使用您用于训练步骤的估算器实例构造一个 RegisterModel 步骤。从 step_train 训练步骤传入 S3ModelArtifacts 属性并指定 ModelPackageGroup。Pipelines 会为您创建此 ModelPackageGroup。

  from sagemaker.model_metrics import MetricsSource, ModelMetrics 
  from sagemaker.workflow.step_collections import RegisterModel
  
  
  model_metrics = ModelMetrics(
      model_statistics=MetricsSource(
          s3_uri="{}/evaluation.json".format(
              step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
          ),
          content_type="application/json"
      )
  )
  step_register = RegisterModel(
      name="AbaloneRegisterModel",
      estimator=xgb_train,
      model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
      content_types=["text/csv"],
      response_types=["text/csv"],
      inference_instances=["ml.t2.medium", "ml.m5.xlarge"],
      transform_instances=["ml.m5.xlarge"],
      model_package_group_name=model_package_group_name,
      approval_status=model_approval_status,
      model_metrics=model_metrics
  )

定义条件步骤以验证模型精度

1. 使用模型评估处理步骤 step_eval 的输出中找到的精度值定义 ConditionLessThanOrEqualTo 条件。使用您在处理步骤中编制索引的属性文件以及相应的JSONPath均方误差值来获取此输出。"mse"

   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=6.0
   )

2. 构造一个 ConditionStep。传入 ConditionEquals 条件，如果条件通过，则将模型包注册和批量转换步骤设置为后续步骤。

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

创建管道

1. 为您的管道定义以下内容：name、parameters 和 steps。名称在 (account, region) 对中必须唯一。

   注意

   一个步骤只能在管道的步骤列表或条件步骤的 if/else 步骤列表中出现一次。不能同时出现在两者中。

   from sagemaker.workflow.pipeline import Pipeline
   
   
   pipeline_name = f"AbalonePipeline"
   pipeline = Pipeline(
       name=pipeline_name,
       parameters=[
           processing_instance_count,
           model_approval_status,
           input_data,
           batch_data,
       ],
       steps=[step_process, step_train, step_eval, step_cond],
   )

2. （可选）检查JSON管道定义以确保其格式正确。

   import json
   
   json.loads(pipeline.definition())