파이프라인 정의

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

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

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

4. 빈 을 선택합니다.

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

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

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

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

9. 오른쪽 사이드바의 나머지 필드를 작성합니다. 이러한 탭의 필드에 대한 자세한 내용은 sagemaker.workflow.steps를 참조하세요ProcessingStep.

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

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

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

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

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. 처리 스크립트를 위한 디렉토리를 생성합니다.

   !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. 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 의 프로세서 인스턴스 방법과 매우 유사합니다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을 지정합니다. 파이프라인은 사용자를 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())