Amazon SageMaker 파이프라인 정의

1. 의 지침에 따라 Studio 콘솔을 엽니다아마존 SageMaker 스튜디오 시작.

2. 왼쪽 탐색 창에서 파이프라인을 선택합니다.

3. 생성(Create)을 선택합니다.

4. 공백을 선택합니다.

5. 왼쪽 사이드바에서 데이터 처리를 선택하고 캔버스로 드래그합니다.

6. 캔버스에서 추가한 프로세스 데이터 단계를 선택합니다.

7. 입력 데이터셋을 추가하려면 오른쪽 사이드바의 데이터 (입력) 에서 추가를 선택하고 데이터세트를 선택합니다.

8. 결과 데이터세트를 저장할 위치를 추가하려면 오른쪽 사이드바의 데이터 (출력) 에서 추가를 선택하고 대상으로 이동합니다.

9. 오른쪽 사이드바의 나머지 필드를 모두 입력합니다. 이 탭의 필드에 대한 자세한 내용은 sagemaker.workflow.steps를 참조하십시오. ProcessingStep.

1. 왼쪽 사이드바에서 Train model을 선택하고 캔버스로 드래그합니다.

2. 캔버스에서 추가한 Train 모델 단계를 선택합니다.

3. 입력 데이터세트를 추가하려면 오른쪽 사이드바의 데이터 (입력) 에서 추가를 선택하고 데이터세트를 선택합니다.

4. 모델 아티팩트를 저장할 위치를 선택하려면 위치 (S3URI) URI 필드에 Amazon S3를 입력하거나 Browse S3를 선택하여 대상 위치로 이동합니다.

5. 오른쪽 사이드바의 나머지 필드를 모두 입력합니다. 이 탭의 필드에 대한 자세한 내용은 sagemaker.workflow.steps를 참조하십시오. TrainingStep.

6. 이전 섹션에서 추가한 프로세스 데이터 단계에서 커서를 클릭하여 모델 학습 단계로 드래그하여 두 단계를 연결하는 가장자리를 만듭니다.

1. 왼쪽 사이드바에서 모델 등록을 선택하고 캔버스로 드래그합니다.

2. 캔버스에서 추가한 모델 등록 단계를 선택합니다.

3. 등록할 모델을 선택하려면 모델 (입력) 에서 추가를 선택합니다.

4. 모델 그룹 생성을 선택하여 새 모델 그룹에 모델을 추가합니다.

5. 오른쪽 사이드바의 나머지 필드를 작성하세요. 이 탭의 필드에 대한 자세한 내용은 sagemaker.workflow.step_collections를 참조하십시오. RegisterModel.

6. 이전 섹션에서 추가한 Train 모델 단계에서 커서를 클릭하고 모델 등록 단계로 드래그하여 두 단계를 연결하는 가장자리를 만듭니다.

1. 왼쪽 사이드바에서 Deploy model (엔드포인트) 을 선택하고 캔버스로 드래그합니다.

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 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 교육 작업 종료 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. 에스티메이터 인스턴스와 의 속성을 TrainingStep 사용하여 생성합니다. ProcessingStep S3Uriof "train" 및 "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. 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처리 S3Uri 단계의 "test" 출력 채널의

   이것은 SageMaker 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 위한 a를 정의하려면

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

a를 정의하여 일괄 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조건을 정의합니다. 처리 단계에서 인덱싱한 속성 파일과 각각의 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())