パイプラインを定義する

1. の指示に従って Studio コンソールを開きますAmazon SageMaker Studio を起動する。

2. 左側のナビゲーションペインで、パイプライン を選択します。

3. [Create] (作成) を選択します。

4. 空白 を選択します。

5. 左側のサイドバーで、データの処理を選択し、キャンバスにドラッグします。

6. キャンバスで、追加したデータ処理ステップを選択します。

7. 入力データセットを追加するには、右側のサイドバーでデータ (入力） の下に追加 を選択し、データセットを選択します。

8. 出力データセットを保存する場所を追加するには、右側のサイドバーでデータ (出力） の下に追加 を選択し、送信先に移動します。

9. 右側のサイドバーの残りのフィールドに入力します。これらのタブのフィールドの詳細については、「sagemaker.workflow.steps」を参照してください。ProcessingStep

1. 左側のサイドバーで、モデルをトレーニングを選択し、キャンバスにドラッグします。

2. キャンバスで、追加したモデルのトレーニングステップを選択します。

3. 入力データセットを追加するには、右側のサイドバーでデータ (入力） の下に追加 を選択し、データセットを選択します。

4. モデルアーティファクトを保存する場所を選択するには、URILocation (S3 ) フィールドに Amazon S3 を入力するか、Browse S3 を選択して送信先の場所に移動します。 URI

5. 右側のサイドバーの残りのフィールドに入力します。これらのタブのフィールドの詳細については、「sagemaker.workflow.steps」を参照してくださいTrainingStep。

6. 前のセクションで追加したデータ処理ステップからモデルをトレーニングステップにカーソルをクリックしてドラッグし、2 つのステップを接続するエッジを作成します。

1. 左側のサイドバーで、モデルを登録を選択し、キャンバスにドラッグします。

2. キャンバスで、追加したモデル登録ステップを選択します。

3. 登録するモデルを選択するには、モデル (入力） で追加を選択します。

4. モデルグループの作成を選択して、モデルを新しいモデルグループに追加します。

5. 右側のサイドバーの残りのフィールドに入力します。これらのタブのフィールドの詳細については、「sagemaker.workflow.step_collections」を参照してくださいRegisterModel。

6. 前のセクションで追加したモデルステップをトレーニングからモデル登録ステップにカーソルをクリックしてドラッグし、2 つのステップを接続するエッジを作成します。

1. 左側のサイドバーで、デプロイモデル (エンドポイント) を選択し、キャンバスにドラッグします。

2. キャンバスで、追加したデプロイモデル (エンドポイント） ステップを選択します。

3. デプロイするモデルを選択するには、モデル (入力） で追加を選択します。

4. エンドポイントの作成ラジオボタンを選択して、新しいエンドポイントを作成します。

5. エンドポイントの名前と説明を入力します。

6. 前のセクションで追加したモデル登録ステップからモデルをデプロイ (エンドポイント） ステップにカーソルをクリックしてドラッグし、2 つのステップを接続するエッジを作成します。

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. モデルを作成したら、バッチ変換用の 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 Python のプロセッサインスタンスの メソッドと非常に似ていますrunSDK。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 は、トレーニングジョブの最後に model.tar.gzの形式でモデルを 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. の推定インスタンスとプロパティを使用して を作成しますTrainingStepProcessingStep。 "train" および "validation" 出力チャネルS3Uriの を に渡します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. /abalone ディレクトリに evaluation.py という名前のファイルを作成します。このスクリプトは、モデル評価を実行するための処理ステップで使用されます。トレーニングされたモデルとテストデータセットを入力として受け取り、分類評価メトリクスを含む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. ProcessingStep の作成に使用される ScriptProcessor のインスタンスを作成します。

   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. プロセッサインスタンス、入力チャネルと出力チャネル、スクリプトを使用して を作成しますProcessingStepevaluation.py。渡します。

   • step_train トレーニングステップの S3ModelArtifactsプロパティ

   • step_process 処理ステップS3Uriの"test"出力チャネルの 。

   これは、 SageMaker Python のプロセッサインスタンスの メソッドと非常に似ていますrunSDK。 

   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. SageMaker モデルを作成します。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 モデル入力を定義します。

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

3. 定義した CreateModelInputおよび SageMaker モデルインスタンスCreateModelStepを使用して を作成します。

   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 ステップリストに 1 回のみ表示できます。また、両方に表示することはできません。

   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())