パイプラインを定義する

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

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

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

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. 処理ステップに渡すプロセッサのインスタンスを作成します。

   from sagemaker.core.processing import Processor
   from sagemaker.core import image_uris
   
   framework_version = "0.23-1"
   
   sklearn_image_uri = image_uris.retrieve(
       framework="sklearn",
       region=region,
       version=framework_version,
       instance_type="ml.m5.xlarge"
   )
   
   sklearn_processor = Processor(
       image_uri=sklearn_image_uri,
       instance_type="ml.m5.xlarge",
       instance_count=processing_instance_count,
       base_job_name="sklearn-abalone-process",
       sagemaker_session=pipeline_session,
       role=role,
   )

2. 処理ステップを作成します。

   from sagemaker.core.processing import ProcessingInput, ProcessingOutput
   from sagemaker.mlops.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 アルゴリズムと入力データセットのトレーニング推定器を設定します。

   from sagemaker.train import ModelTrainer
   from sagemaker.train.configs import Compute, OutputDataConfig
   from sagemaker.core import image_uris
   
   image_uri = image_uris.retrieve(
       framework="xgboost",
       region=region,
       version="1.0-1",
       py_version="py3",
       instance_type="ml.m5.xlarge"
   )
   xgb_train = ModelTrainer(
       training_image=image_uri,
       compute=Compute(instance_type="ml.m5.xlarge", instance_count=1),
       output_data_config=OutputDataConfig(s3_output_path=model_path),
       hyperparameters={
           "objective": "reg:linear",
           "num_round": "50",
           "max_depth": "5",
           "eta": "0.2",
           "gamma": "4",
           "min_child_weight": "6",
           "subsample": "0.7",
           "silent": "0"
       },
       sagemaker_session=pipeline_session,
       role=role,
   )

3. のトレーニングインスタンスとプロパティTrainingStepを使用して を作成しますProcessingStep。

   from sagemaker.train.configs import InputData
   from sagemaker.mlops.workflow.steps import TrainingStep
   
   train_args = xgb_train.train(
       input_data_config=[
           InputData(
               channel_name="train",
               data_source=step_process.properties.ProcessingOutputConfig.Outputs[
                   "train"
               ].S3Output.S3Uri,
               content_type="text/csv"
           ),
           InputData(
               channel_name="validation",
               data_source=step_process.properties.ProcessingOutputConfig.Outputs[
                   "validation"
               ].S3Output.S3Uri,
               content_type="text/csv"
           )
       ],
   )
   
   step_train = TrainingStep(name="AbaloneTrain", step_args=train_args)

モデル評価の処理ステップを定義するには

1. 評価Processor用の のインスタンスを作成します。

   from sagemaker.core.processing import Processor
   
   script_eval = Processor(
       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,
   )

2. プロセッサインスタンス、入力チャネルと出力チャネル、スクリプトProcessingStepを使用して を作成しますevaluation.py。

   from sagemaker.core.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 を定義するには

• SageMaker AI モデルとモデルステップを作成します。

  from sagemaker.serve import ModelBuilder
  from sagemaker.mlops.workflow.model_step import ModelStep
  
  model_builder = ModelBuilder(
      image_uri=image_uri,
      s3_model_data_url=step_train.properties.ModelArtifacts.S3ModelArtifacts,
      role_arn=role,
      sagemaker_session=pipeline_session,
  )
  
  step_create_model = ModelStep(
      name="AbaloneCreateModel",
      step_args=model_builder.build(),
  )

バッチ変換を実行するための TransformStep を定義するには

• トランスフォーマーインスタンスと を作成しますTransformStep。

  from sagemaker.core.transformer import Transformer
  from sagemaker.mlops.workflow.steps import TransformStep
  
  transformer = Transformer(
      model_name=step_create_model.properties.ModelName,
      instance_type="ml.m5.xlarge",
      instance_count=1,
      output_path=f"s3://{default_bucket}/AbaloneTransform",
      sagemaker_session=pipeline_session,
  )
  
  transform_args = transformer.transform(data=batch_data)
  
  step_transform = TransformStep(
      name="AbaloneTransform",
      step_args=transform_args,
  )

モデル登録ステップを定義してモデルパッケージを作成するには

• モデル登録ステップを作成します。

  from sagemaker.core.model_metrics import MetricsSource, ModelMetrics 
  from sagemaker.mlops.workflow.model_step import ModelStep
  from sagemaker.serve import ModelBuilder
  
  model_metrics = ModelMetrics(
      model_statistics=MetricsSource(
          s3_uri="{}/evaluation.json".format(
              step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
          ),
          content_type="application/json"
      )
  )
  
  model_builder = ModelBuilder(
      image_uri=image_uri,
      s3_model_data_url=step_train.properties.ModelArtifacts.S3ModelArtifacts,
      role_arn=role,
      sagemaker_session=pipeline_session,
  )
  
  register_args = model_builder.register(
      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
  )
  
  step_register = ModelStep(
      name="AbaloneRegisterModel",
      step_args=register_args,
  )

モデルの精度を検証するための条件ステップを定義するには

• 条件を定義し、 を構築しますConditionStep。

  from sagemaker.core.workflow.conditions import ConditionLessThanOrEqualTo
  from sagemaker.mlops.workflow.condition_step import ConditionStep
  from sagemaker.core.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
  )
  
  step_cond = ConditionStep(
      name="AbaloneMSECond",
      conditions=[cond_lte],
      if_steps=[step_register, step_create_model, step_transform],
      else_steps=[], 
  )

パイプラインを作成するには

1. パイプラインの name、parameters、steps を定義します。

   from sagemaker.mlops.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())