パイプラインを定義する

1. 「Amazon SageMaker Studio を起動する」の手順に従って、Studio コンソールを開きます。

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

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

4. [指定なし] をクリックします。

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

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

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

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

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

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

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

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

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

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 AI Python SDK のプロセッサインスタンスの メソッドと非常によく似ていますrun。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 は、トレーニングジョブの終了時にモデルを の形式で Amazon S3 にアップロードします。model.tar.gz

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

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

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

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

   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 AI モデルを作成します。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. 定義した CreateModelInputと SageMaker AI モデルインスタンス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 を作成します。パイプラインは、この 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 条件を定義します。この出力を取得するには、処理ステップでインデックス付けしたプロパティファイルと、平均二乗誤差値 "mse" のそれぞれの JSONPath を使用します。

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