파이프라인 정의

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. 이전 섹션에서 추가한 프로세스 데이터 단계에서 커서를 클릭하고 모델 훈련 단계로 드래그하여 두 단계를 연결하는 엣지를 만듭니다.

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. 처리 단계에 전달할 프로세서의 인스턴스를 생성합니다.

   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],
   )

배치 변환을 위한 모델 생성 단계를 정의하려면

• 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.

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