단계 추가

Pipelines 드래그 앤 드롭 UI에서 코드 실행 단계를 사용하여 자체 코드를 파이프라인 단계로 실행할 수 있습니다. 파이프라인의 일부로 실행할 Python 함수, 스크립트 또는 노트북을 업로드할 수 있습니다. 고급 SageMaker AI 기능 또는 기타 AWS 서비스를 활용하는 사용자 지정 ML 작업을 오케스트레이션하려면이 단계를 사용해야 합니다.

코드 실행 단계는 Amazon SageMaker AI용 기본 Amazon S3 버킷에 파일을 업로드합니다. Amazon SageMaker 이 버킷에는 필요한 교차 오리진 리소스 공유(CORS) 권한 세트가 없을 수 있습니다. CORS 권한 구성에 대한 자세한 내용은 입력 이미지 데이터에 대한 CORS 요구 사항 섹션을 참조하세요.

코드 실행 단계에서는 Amazon SageMaker 훈련 작업을 사용하여 코드를 실행합니다. IAM 역할에 sagemaker:DescribeTrainingJob 및 sagemaker:CreateTrainingJob API 권한이 있어야 합니다. Amazon SageMaker AI에 필요한 모든 권한과 이를 설정하는 방법에 대한 자세한 내용은 섹션을 참조하세요Amazon SageMaker AI API 권한: 작업, 권한 및 리소스 참조.

파이프라인 디자이너를 사용하여 파이프라인에 코드 실행 단계를 추가하려면 다음을 수행합니다.

처리 단계를 사용하여 데이터 처리를 위한 처리 작업을 생성합니다. 작업 처리에 대한 자세한 내용은 데이터 처리 및 모델 평가를 참조하세요.

훈련 단계를 사용하여 모델을 훈련시키는 훈련 작업을 생성합니다. 훈련 작업에 대한 자세한 내용은 Amazon SageMaker AI로 모델 훈련을 참조하세요.

훈련 단계에는 훈련 및 검증 데이터 입력뿐만 아니라 예측기가 필요합니다.

튜닝 단계를 사용하여 하이퍼파라미터 최적화(HPO) 라고도 하는 하이퍼파라미터 튜닝 작업을 생성합니다. 하이퍼파라미터 조정 작업은 여러 개의 훈련 작업을 실행하며 각 작업은 모델 버전을 생성합니다. 하이퍼파라미터 튜닝에 대한 자세한 내용은 SageMaker AI를 사용한 자동 모델 튜닝섹션을 참조하세요.

튜닝 작업은 파이프라인에 대한 SageMaker AI 실험과 연결되어 있으며 훈련 작업은 시도로 생성됩니다. 자세한 내용은 실험 통합 단원을 참조하십시오.

튜닝 단계에는 하이퍼파라미터 튜너와 훈련 입력이 필요합니다. HyperparameterTuner의 warm_start_config파라미터를 지정하여 이전 튜닝 작업을 다시 훈련할 수 있습니다. 하이퍼파라미터 튜닝 및 부팅에 대한 자세한 내용은 웜 스타트 하이퍼파라미터 튜닝 작업 실행섹션을 참조하세요.

SageMaker.Workflow.steps.TuningStep 클래스의 get_top_model_s3_uri 메서드를 사용하여 성능이 가장 우수한 모델 버전 중 하나에서 모델 아티팩트를 가져올 수 있습니다. SageMaker AI 파이프라인에서 튜닝 단계를 사용하는 방법을 보여주는 노트북은 sagemaker-pipelines-tuning-step.ipynb를 참조하세요.

다음 예제는 TuningStep정의를 생성하는 방법을 보여줍니다.

from sagemaker.workflow.pipeline_context import PipelineSession

from sagemaker.tuner import HyperparameterTuner
from sagemaker.inputs import TrainingInput
from sagemaker.workflow.steps import TuningStep

tuner = HyperparameterTuner(..., sagemaker_session=PipelineSession())
    
step_tuning = TuningStep(
    name = "HPTuning",
    step_args = tuner.fit(inputs=TrainingInput(s3_data="s3://amzn-s3-demo-bucket/my-data"))
)

최적 모델 버전 가져오기

다음 예제에서는 get_top_model_s3_uri메서드를 사용하여 튜닝 작업에서 최적 모델 버전을 가져오는 방법을 보여줍니다. 최고 성능 상위 50개 버전은 HyperParameterTuningJobObjective에 따라 순위가 매겨집니다. top_k 인수는 버전에 대한 인덱스로, top_k=0은 성능이 가장 좋은 버전이고 top_k=49는 성능이 가장 낮은 버전입니다.

best_model = Model(
    image_uri=image_uri,
    model_data=step_tuning.get_top_model_s3_uri(
        top_k=0,
        s3_bucket=sagemaker_session.default_bucket()
    ),
    ...
)

튜닝 단계 요구 사항에 대한 자세한 내용은 SageMaker.Workflow.steps.TuningStep 설명서를 참조하세요.

미세 조정은 새 데이터세트에 대해 Amazon SageMaker JumpStart의 사전 훈련된 파운데이션 모델을 훈련합니다. 전이 학습이라고도 하는 이 프로세스를 통해 더 작은 데이터세트를 사용하고 훈련 시간을 단축하여 정확한 모델을 생성할 수 있습니다. 모델을 미세 조정할 때 기본 데이터세트를 사용하거나 자체 데이터를 선택할 수 있습니다. JumpStart에서 파운데이션 모델을 미세 조정하는 방법에 대한 자세한 내용은 모델 미세 조정 섹션을 참조하세요.

미세 조정 단계에서는 Amazon SageMaker 훈련 작업을 사용하여 모델을 사용자 지정합니다. IAM 역할에 파이프라인에서 미세 조정 작업을 실행할 수 있는 sagemaker:DescribeTrainingJob 및 sagemaker:CreateTrainingJob API 권한이 있어야 합니다. Amazon SageMaker AI에 필요한 권한과 이를 설정하는 방법에 대한 자세한 내용은 섹션을 참조하세요Amazon SageMaker AI API 권한: 작업, 권한 및 리소스 참조.

드래그 앤 드롭 편집기를 사용하여 파이프라인에 미세 조정 모델 단계를 추가하려면 다음 단계를 따르세요.

AutoML API를 사용하여 모델을 자동으로 학습시키는 AutoML 작업을 생성할 수 있습니다. AutoML 작업에 대한 자세한 내용은 Amazon SageMaker Autopilot을 사용한 모델 개발 자동화를 참조하세요.

다음 예제는 AutoMLStep을 사용하여 정의를 생성하는 방법을 보여줍니다.

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.workflow.automl_step import AutoMLStep

pipeline_session = PipelineSession()

auto_ml = AutoML(...,
    role="<role>",
    target_attribute_name="my_target_attribute_name",
    mode="ENSEMBLING",
    sagemaker_session=pipeline_session) 

input_training = AutoMLInput(
    inputs="s3://amzn-s3-demo-bucket/my-training-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="training",
)
input_validation = AutoMLInput(
    inputs="s3://amzn-s3-demo-bucket/my-validation-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="validation",
)

step_args = auto_ml.fit(
    inputs=[input_training, input_validation]
)

step_automl = AutoMLStep(
    name="AutoMLStep",
    step_args=step_args,
)     

최적 모델 버전 가져오기

AutoML 단계에서는 여러 모델 후보를 자동으로 학습시킵니다. 다음과 같이 get_best_auto_ml_model 메서드를 사용하여 AutoML 작업에서 최적의 목표 지표가 있는 모델을 가져옵니다. 또한 IAM role을 사용하여 모델 아티팩트에 액세스해야 합니다.

best_model = step_automl.get_best_auto_ml_model(role=<role>)

자세한 내용은 SageMaker Python SDK의 AutoML 단계를 참조하세요.

ModelStep를 사용하여 SageMaker AI 모델을 생성하거나 등록합니다. ModelStep 요구 사항에 대한 자세한 내용은 SageMaker.Workflow.Model_Step.ModelStep 설명서를 참조하세요.

모델 생성

ModelStep를 사용하여 SageMaker AI 모델을 생성할 수 있습니다. ModelStep 에는 모델을 생성하는 데 사용해야 하는 SageMaker AI 인스턴스 유형에 대한 모델 아티팩트와 정보가 필요합니다. SageMaker AI 모델에 대한 자세한 내용은 Amazon SageMaker AI로 모델 훈련을 참조하세요.

다음 예제는 ModelStep정의를 생성하는 방법을 보여줍니다.

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.model import Model
from sagemaker.workflow.model_step import ModelStep

step_train = TrainingStep(...)
model = Model(
    image_uri=pytorch_estimator.training_image_uri(),
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    sagemaker_session=PipelineSession(),
    role=role,
)

step_model_create = ModelStep(
   name="MyModelCreationStep",
   step_args=model.create(instance_type="ml.m5.xlarge"),
)

모델 등록

를 사용하여 Amazon SageMaker 모델 레지스트리ModelStep에 sagemaker.model.Model 또는 sagemaker.pipeline.PipelineModel를 등록할 수 있습니다. PipelineModel은 추론 파이프라인은 데이터에 대해 추론 요청을 처리하는 컨테이너로 이루어진 선형 시퀀스로 구성된 모델입니다. 모델을 등록하는 자세한 방법은 Model Registry를 사용한 모델 등록 배포섹션을 참조하세요.

다음 예제에서는 PipelineModel을 등록하는 ModelStep생성 방법을 보여 줍니다.

import time

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.sklearn import SKLearnModel
from sagemaker.xgboost import XGBoostModel

pipeline_session = PipelineSession()

code_location = 's3://{0}/{1}/code'.format(bucket_name, prefix)

sklearn_model = SKLearnModel(
   model_data=processing_step.properties.ProcessingOutputConfig.Outputs['model'].S3Output.S3Uri,
   entry_point='inference.py',
   source_dir='sklearn_source_dir/',
   code_location=code_location,
   framework_version='1.0-1',
   role=role,
   sagemaker_session=pipeline_session,
   py_version='py3'
)

xgboost_model = XGBoostModel(
   model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
   entry_point='inference.py',
   source_dir='xgboost_source_dir/',
   code_location=code_location,
   framework_version='0.90-2',
   py_version='py3',
   sagemaker_session=pipeline_session,
   role=role
)

from sagemaker.workflow.model_step import ModelStep
from sagemaker import PipelineModel

pipeline_model = PipelineModel(
   models=[sklearn_model, xgboost_model],
   role=role,sagemaker_session=pipeline_session,
)

register_model_step_args = pipeline_model.register(
    content_types=["application/json"],
   response_types=["application/json"],
   inference_instances=["ml.t2.medium", "ml.m5.xlarge"],
   transform_instances=["ml.m5.xlarge"],
   model_package_group_name='sipgroup',
)

step_model_registration = ModelStep(
   name="AbaloneRegisterModel",
   step_args=register_model_step_args,
)

모델 생성 단계를 사용하여 SageMaker AI 모델을 생성합니다. SageMaker AI 모델에 대한 자세한 내용은 섹션을 참조하세요Amazon SageMaker를 사용한 모델 훈련.

모델 생성 단계에는 모델을 생성하는 데 사용해야 하는 SageMaker AI 인스턴스 유형에 대한 모델 아티팩트와 정보가 필요합니다. 다음 예시는 모델 만들기 단계 정의를 만드는 방법을 보여줍니다. 모델 만들기 단계 요구 사항에 대한 자세한 내용은 sagemaker.workflow.steps.CreateModelStep 설명서를 참조하세요.

모델 등록 단계에서는 모델을 SageMaker Model Registry에 등록합니다.

파이프라인 디자이너에서 모델 배포(엔드포인트) 단계를 사용하여 모델을 엔드포인트에 배포합니다. 새 엔드포인트를 만들거나 기존 엔드포인트를 사용할 수 있습니다. 실시간 추론은 실시간, 대화형, 짧은 지연 시간이 요구되는 추론 워크로드에 적합합니다. 모델을 SageMaker AI 호스팅 서비스에 배포하고 추론에 사용할 수 있는 실시간 엔드포인트를 얻을 수 있습니다. 이러한 엔드포인트는 완전 관리형이며 오토 스케일링을 지원합니다. SageMaker AI의 실시간 추론에 대한 자세한 내용은 섹션을 참조하세요실시간 추론.

파이프라인에 모델 배포 단계를 추가하기 전에 IAM 역할에 다음 권한이 있는지 확인합니다.

SageMaker AI에 필요한 모든 권한과 이를 설정하는 방법에 대한 자세한 내용은 섹션을 참조하세요Amazon SageMaker AI API 권한: 작업, 권한 및 리소스 참조.

드래그 앤 드롭 편집기에서 파이프라인에 모델 배포 단계를 추가하려면 다음 단계를 완료하세요.

배치 변환을 위한 변환 단계를 사용하여 전체 데이터세트에 대해 추론을 실행합니다. 배치 변환에 대한 자세한 정보는 추론 파이프라인을 사용하여 배치 변환섹션을 참조하세요.

변환 단계에는 배치 변환을 실행하는 데 사용할 변환기와 데이터가 필요합니다. 다음 예시는 변환 단계 정의를 만드는 방법을 보여줍니다. 변환 단계 요구 사항에 대한 자세한 내용은 sagemaker.workflow.steps.TransformStep 문서를 참조하세요.

조건 단계를 사용하여 단계 속성의 조건을 평가하여 파이프라인에서 다음에 취해야 할 조치를 평가합니다.

조건 단계에는 다음이 필요합니다.

Callback 단계를 사용하여 Amazon SageMaker Pipelines에서 직접 제공하지 않는 추가 프로세스와 AWS 서비스를 워크플로에 추가합니다. Callback 단계가 실행되면 다음 절차가 수행됩니다.

Callback 단계 요구 사항에 대한 자세한 내용은 SageMaker.Workflow.Callback_step.CallbackStep 설명서를 참조하세요. 전체 솔루션을 보려면 콜백 단계를 사용하여 사용자 지정 단계를 포함하도록 SageMaker Pipelines 확장을 참조하세요.

다음 샘플은 이전 절차의 구현을 보여줍니다.

from sagemaker.workflow.callback_step import CallbackStep

step_callback = CallbackStep(
    name="MyCallbackStep",
    sqs_queue_url="https://sqs.us-east-2.amazonaws.com/012345678901/MyCallbackQueue",
    inputs={...},
    outputs=[...]
)

callback_handler_code = '
    import boto3
    import json

    def handler(event, context):
        sagemaker_client=boto3.client("sagemaker")

        for record in event["Records"]:
            payload=json.loads(record["body"])
            token=payload["token"]

            # Custom processing

            # Call SageMaker AI to complete the step
            sagemaker_client.send_pipeline_execution_step_success(
                CallbackToken=token,
                OutputParameters={...}
            )
'

중지 동작

Callback 단계가 실행되는 동안에는 파이프라인 프로세스가 중지되지 않습니다.

실행 중인 Callback 단계가 있는 파이프라인 프로세스에서 StopPipelineExecution을 직접 호출하면 Pipelines은 SQS 대기열에 Amazon SQS 메시지를 보냅니다. SQS 메시지 본문에는 Stopping으로 설정된 상태 필드가 포함되어 있습니다. 다음은 예제 SQS 메시지 본문을 보여줍니다.

{
  "token": "26vcYbeWsZ",
  "pipelineExecutionArn": "arn:aws:sagemaker:us-east-2:012345678901:pipeline/callback-pipeline/execution/7pinimwddh3a",
  "arguments": {
    "number": 5,
    "stringArg": "some-arg",
    "inputData": "s3://sagemaker-us-west-2-012345678901/abalone/abalone-dataset.csv"
  },
  "status": "Stopping"
}

메시지를 수신한 후 필요한 조치(예: 리소스 정리)를 수행할 수 있는 로직을 Amazon SQS 메시지 소비자에게 추가해야 합니다. 그런 다음 SendPipelineExecutionStepSuccess 또는 SendPipelineExecutionStepFailure에 직접 호출을 추가합니다.

Pipelines은 이러한 호출 중 하나를 수신한 경우에만 파이프라인 프로세스를 중지합니다.

Lambda 단계를 사용하여 AWS Lambda 함수를 실행합니다. 기존 Lambda 함수를 실행하거나 SageMaker AI가 새 Lambda 함수를 생성하고 실행할 수 있습니다. SageMaker AI 파이프라인에서 Lambda 단계를 사용하는 방법을 보여주는 노트북은 sagemaker-pipelines-lambda-step.ipynb를 참조하세요.

SageMaker AI는 Lambda 함수를 생성, 업데이트, 호출 및 삭제할 수 있는 sagemaker.lambda_helper.Lambda 클래스를 제공합니다. Lambda에는 다음 서명이 있습니다.

Lambda(
    function_arn,       # Only required argument to invoke an existing Lambda function

    # The following arguments are required to create a Lambda function:
    function_name,
    execution_role_arn,
    zipped_code_dir,    # Specify either zipped_code_dir and s3_bucket, OR script
    s3_bucket,          # S3 bucket where zipped_code_dir is uploaded
    script,             # Path of Lambda function script
    handler,            # Lambda handler specified as "lambda_script.lambda_handler"
    timeout,            # Maximum time the Lambda function can run before the lambda step fails
    ...
)

Sagemaker.workflow.lambda_step.LambdaStep 클래스에는 Lambda유형의 lambda_func인수가 있습니다. 기존 Lambda 함수를 호출하려면 함수의 Amazon 리소스 이름(ARN)을 function_arn에 제공하기만 하면 됩니다. function_arn에 대한 값을 제공하지 않는 경우 handler와 다음 중 하나를 지정해야 합니다.

다음 예제는 기존 Lambda 함수를 호출하는 Lambda단계 정의를 생성하는 방법을 보여줍니다.

from sagemaker.workflow.lambda_step import LambdaStep
from sagemaker.lambda_helper import Lambda

step_lambda = LambdaStep(
    name="ProcessingLambda",
    lambda_func=Lambda(
        function_arn="arn:aws:lambda:us-west-2:012345678910:function:split-dataset-lambda"
    ),
    inputs={
        s3_bucket = s3_bucket,
        data_file = data_file
    },
    outputs=[
        "train_file", "test_file"
    ]
)

다음 예제는 Lambda 함수 스크립트를 사용하여 Lambda 함수를 생성하고 호출하는 Lambda단계 정의를 생성하는 방법을 보여줍니다.

from sagemaker.workflow.lambda_step import LambdaStep
from sagemaker.lambda_helper import Lambda

step_lambda = LambdaStep(
    name="ProcessingLambda",
    lambda_func=Lambda(
      function_name="split-dataset-lambda",
      execution_role_arn=execution_role_arn,
      script="lambda_script.py",
      handler="lambda_script.lambda_handler",
      ...
    ),
    inputs={
        s3_bucket = s3_bucket,
        data_file = data_file
    },
    outputs=[
        "train_file", "test_file"
    ]
)

입력 및 출력

Lambda 함수에 입력 또는 출력이 있는 경우 해당 입력 또는 출력도 Lambda단계에서 정의해야 합니다.

Lambda 단계 정의 시, inputs은 키-값 쌍의 딕셔너리여야 합니다. inputs 사전의 각 값은 프리미티브 유형(문자열, 정수 또는 부동)이어야 합니다. 중첩된 개체는 지원되지 않습니다. 정의되지 않은 상태로 두면 inputs기본값은 None입니다.

outputs 값은 키 목록이어야 합니다. 이 키는 Lambda함수 출력에 정의된 사전을 참조합니다. inputs과 마찬가지로 이러한 키는 프리미티브 유형이어야 하며 중첩된 객체는 지원되지 않습니다.

타임아웃 및 중지 동작

Lambda 클래스에는 Lambda 함수를 실행할 수 있는 최대 시간을 지정하는 timeout인수가 있습니다. 기본값은 120초이며 최대값은 10분입니다. 제한 시간에 도달했을 때 Lambda 함수가 실행 중이면 Lambda 단계는 실패하지만 Lambda 함수는 계속 실행됩니다.

Lambda 단계에서 호출한 Lambda 함수를 중지할 수 없기 때문에 Lambda 단계가 실행되는 동안에는 파이프라인 프로세스를 중지할 수 없습니다. Lambda 함수가 실행되는 동안 프로세스를 중지하면 파이프라인은 함수가 완료될 때까지 또는 제한 시간에 도달할 때까지 기다립니다. 둘 중 어느 상황이 먼저 발생하는지에 따라 달라집니다. 그런 다음 프로세스가 중지됩니다. Lambda 함수가 완료되면 파이프라인 프로세스 상태는 Stopped입니다. 제한 시간에 도달하면 파이프라인 프로세스 상태는 Failed입니다.

이 ClarifyCheck단계를 사용하여 바이어스 분석 및 모델 설명 가능성에 대해 이전 기준 대비 기준 드리프트 검사를 수행할 수 있습니다. 그런 다음 model.register()메서드로 기준을 생성 및 등록하고 해당 메서드의 결과를 step_args을 사용하여 모델 단계에 전달할 수 있습니다. 이러한 드리프트 검사 기준은 Amazon SageMaker Model Monitor가 모델 엔드포인트에 사용할 수 있습니다. 따라서 기준 제안을 별도로 수행할 필요가 없습니다.

ClarifyCheck 단계에서는 모델 레지스트리에서 드리프트 검사의 기준을 가져올 수도 있습니다. ClarifyCheck 단계에서는 SageMaker Clarify 사전 빌드된 컨테이너를 사용합니다. 이 컨테이너는 제약 조건 제안 및 주어진 기준에 대한 제약 조건 검증을 비롯한 다양한 모델 모니터링 기능을 제공합니다. 자세한 내용은 사전 구축된 SageMaker Clarify 컨테이너 단원을 참조하십시오.

ClarifyCheck 단계 구성

파이프라인에서 사용할 때마다 사용 가능한 검사 유형 중 하나만을 수행하도록 ClarifyCheck단계를 구성할 수 있습니다.

그러기 위해 clarify_check_config 파라미터를 다음 검사 유형 값 중 하나로 설정합니다.

이 ClarifyCheck 단계에서는 SageMaker AI Clarify 사전 구축 컨테이너를 실행하는 처리 작업을 시작하고 검사 및 처리 작업을 위한 전용 구성이 필요합니다. ClarifyCheckConfig 및 CheckJobConfig는 이러한 구성에 대한 헬퍼 함수입니다. 이러한 도우미 함수는 SageMaker Clarify 처리 작업이 모델 편향, 데이터 편향 또는 모델 설명 가능성을 검사하기 위해 계산하는 방식에 부합합니다. 자세한 내용은 편향 분석 및 설명 가능성을 위한 SageMaker Clarify 처리 작업 실행 단원을 참조하십시오.

드리프트 검사의 단계 동작 제어

ClarifyCheck 단계의 동작을 제어하려면 다음 두 개의 부울 플래그가 필요합니다.

자세한 내용은 Amazon SageMaker Pipelines의 명확화 검사 및 품질 검사 단계를 사용한 기준 계산, 드리프트 감지 및 수명 주기섹션을 참조하세요.

기준 작업

선택적으로 model_package_group_name을 지정하여 기존 기준을 찾을 수 있습니다. 그런 다음 ClarifyCheck 단계는 모델 패키지 그룹의 승인된 최신 모델 패키지에서 DriftCheckBaselines을 가져옵니다.

또는 supplied_baseline_constraints파라미터를 통해 이전 기준을 제공할 수 있습니다. model_package_group_name, supplied_baseline_constraints를 모두 지정하는 경우 ClarifyCheck단계에서는 supplied_baseline_constraints파라미터로 지정된 기준을 사용합니다.

ClarifyCheck 단계 요구 사항 사용에 대한 자세한 내용은 Python용 Amazon SageMaker AI SageMaker AI SDK의 sagemaker.workflow.steps.ClarifyCheckStep을 참조하세요. Amazon SageMaker SageMaker Pipelines에서 ClarifyCheck 단계를 사용하는 방법을 보여주는 Amazon SageMaker Studio Classic 노트북은 sagemaker-pipeline-model-monitor-clarify-steps.ipynb를 참조하세요.

from sagemaker.workflow.check_job_config import CheckJobConfig
from sagemaker.workflow.clarify_check_step import DataBiasCheckConfig, ClarifyCheckStep
from sagemaker.workflow.execution_variables import ExecutionVariables

check_job_config = CheckJobConfig(
    role=role,
    instance_count=1,
    instance_type="ml.c5.xlarge",
    volume_size_in_gb=120,
    sagemaker_session=sagemaker_session,
)

data_bias_data_config = DataConfig(
    s3_data_input_path=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
    s3_output_path=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'databiascheckstep']),
    label=0,
    dataset_type="text/csv",
    s3_analysis_config_output_path=data_bias_analysis_cfg_output_path,
)

data_bias_config = BiasConfig(
    label_values_or_threshold=[15.0], facet_name=[8], facet_values_or_threshold=[[0.5]]  
)

data_bias_check_config = DataBiasCheckConfig(
    data_config=data_bias_data_config,
    data_bias_config=data_bias_config,
)h

data_bias_check_step = ClarifyCheckStep(
    name="DataBiasCheckStep",
    clarify_check_config=data_bias_check_config,
    check_job_config=check_job_config,
    skip_check=False,
    register_new_baseline=False
   supplied_baseline_constraints="s3://sagemaker-us-west-2-111122223333/baseline/analysis.json",
    model_package_group_name="MyModelPackageGroup"
)

QualityCheck 단계를 사용하여 파이프라인의 데이터 품질 또는 모델 품질에 대한 이전 기준과 비교하여 기준 제안 및 드리프트 검사를 수행합니다. 그런 다음 model.register() 메서드로 기준을 생성 및 등록하고 해당 메서드의 결과를 step_args를 사용하여 모델 단계에 전달할 수 있습니다.

Model Monitor는 이러한 기준을 모델 엔드포인트의 드리프트 검사에 사용할 수 있으므로 기준 제안을 별도로 수행할 필요가 없습니다. QualityCheck 단계에서는 모델 레지스트리에서 드리프트 검사의 기준을 가져올 수도 있습니다. 이 QualityCheck 단계에서는 Amazon SageMaker AI 모델 모니터 사전 빌드된 컨테이너를 활용합니다. 이 컨테이너에는 제약 조건 제안, 통계 생성, 주어진 기준에 대한 제약 조건 검증을 비롯한 다양한 모델 모니터링 기능이 있습니다. 자세한 내용은 Amazon SageMaker Model Monitor 사전 구축 컨테이너 단원을 참조하십시오.

QualityCheck 단계 구성

파이프라인에서 사용할 때마다 사용 가능한 검사 유형 중 하나만을 실행하도록 QualityCheck 단계를 구성할 수 있습니다.

다음 검사 유형 값 중 하나로 quality_check_config파라미터를 설정하여 이 작업을 수행합니다.

QualityCheck 단계에서는 Model Monitor 사전 빌드된 컨테이너를 실행하는 처리 작업을 시작하며 검사 및 처리 작업을 위한 전용 구성이 필요합니다. QualityCheckConfig 및 CheckJobConfig는 이러한 구성에 대한 도우미 함수입니다. 이러한 도우미 함수는 Model Monitor가 모델 품질 또는 데이터 품질 모니터링을 위한 기준을 만드는 방식에 부합합니다. 모델 모니터 기준 제안에 대한 자세한 내용은 기준 생성, 모델 품질 기준 만들기섹션을 참조하세요.

드리프트 검사의 단계 동작 제어

QualityCheck 단계의 동작을 제어하려면 다음 두 개의 부울 플래그가 필요합니다.

자세한 내용은 Amazon SageMaker Pipelines의 명확화 검사 및 품질 검사 단계를 사용한 기준 계산, 드리프트 감지 및 수명 주기섹션을 참조하세요.

기준 작업

supplied_baseline_statistics 및 supplied_baseline_constraints 파라미터를 통해 직접 이전 기준을 지정할 수 있습니다. 또한 model_package_group_name을 지정할 수 있으며, QualityCheck 단계는 모델 패키지 그룹의 승인된 최신 모델 패키지에서 DriftCheckBaselines을 가져옵니다.

다음을 지정하면 QualityCheck 단계는 QualityCheck 단계의 검사 유형에서 supplied_baseline_constraints 및 supplied_baseline_statistics으로 지정된 기준을 사용합니다.

QualityCheck 단계 요구 사항 사용에 대한 자세한 내용은 Python용 Amazon SageMaker AI SageMaker AI SDK의 sagemaker.workflow.steps.QualityCheckStep을 참조하세요. Amazon SageMaker SageMaker Pipelines에서 QualityCheck 단계를 사용하는 방법을 보여주는 Amazon SageMaker Studio Classic 노트북은 sagemaker-pipeline-model-monitor-clarify-steps.ipynb를 참조하세요.

from sagemaker.workflow.check_job_config import CheckJobConfig
from sagemaker.workflow.quality_check_step import DataQualityCheckConfig, QualityCheckStep
from sagemaker.workflow.execution_variables import ExecutionVariables

check_job_config = CheckJobConfig(
    role=role,
    instance_count=1,
    instance_type="ml.c5.xlarge",
    volume_size_in_gb=120,
    sagemaker_session=sagemaker_session,
)

data_quality_check_config = DataQualityCheckConfig(
    baseline_dataset=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
    dataset_format=DatasetFormat.csv(header=False, output_columns_position="START"),
    output_s3_uri=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'dataqualitycheckstep'])
)

data_quality_check_step = QualityCheckStep(
    name="DataQualityCheckStep",
    skip_check=False,
    register_new_baseline=False,
    quality_check_config=data_quality_check_config,
    check_job_config=check_job_config,
    supplied_baseline_statistics="s3://sagemaker-us-west-2-555555555555/baseline/statistics.json",
    supplied_baseline_constraints="s3://sagemaker-us-west-2-555555555555/baseline/constraints.json",
    model_package_group_name="MyModelPackageGroup"
)

Amazon SageMaker Pipelines EMR 단계를 사용하여 다음을 수행합니다.

Amazon EMR에 대한 자세한 내용은 Amazon EMR 시작하기를 참조하세요.

EMR 단계를 수행하려면 Amazon EMR 클러스터에서 사용하는 JAR 파일의 위치와 전달할 인수가 EMRStepConfig에 포함되어야 합니다. 실행 중인 EMR 클러스터에서 단계를 실행하려는 경우 Amazon EMR 클러스터 ID도 제공합니다. 클러스터 구성을 전달하여 클러스터가 만들고 관리하고 종료하는 클러스터에서 EMR 단계를 실행할 수도 있습니다. 다음 섹션에는 두 가지 방법을 모두 보여주는 샘플 노트북으로 연결되는 링크와 예제가 포함되어 있습니다.

실행 중인 Amazon EMR 클러스터에서 새 작업 시작

실행 중인 Amazon EMR 클러스터에서 새 작업을 시작하려면 클러스터 ID를 EMRStep의 cluster_id 인수에 문자열로 전달합니다. 다음 예에서는 이 절차를 보여 줍니다.

from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_config = EMRStepConfig(
    jar="jar-location", # required, path to jar file used
    args=["--verbose", "--force"], # optional list of arguments to pass to the jar
    main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest 
    properties=[ # optional list of Java properties that are set when the step runs
    {
        "key": "mapred.tasktracker.map.tasks.maximum",
        "value": "2"
    },
    {
        "key": "mapreduce.map.sort.spill.percent",
        "value": "0.90"
   },
   {
       "key": "mapreduce.tasktracker.reduce.tasks.maximum",
       "value": "5"
    }
  ]
)

step_emr = EMRStep (
    name="EMRSampleStep", # required
    cluster_id="j-1ABCDEFG2HIJK", # include cluster_id to use a running cluster
    step_config=emr_config, # required
    display_name="My EMR Step",
    description="Pipeline step to execute EMR job"
)

전체 예시를 안내하는 샘플 노트북은 Pipelines EMR Step With Running EMR Cluster를 참조하세요.

실행 중인 Amazon EMR 클러스터에서 새 작업 시작

EMRStep이 사용자를 대신해 만드는 새 클러스터에서 새 작업을 시작하려면 클러스터 구성을 딕셔너리로 제공합니다. 딕셔너리는 RunJobFlow 요청과 구조가 동일해야 합니다. 하지만 클러스터 구성에 다음 필드를 포함시키지 마세요.

다른 모든 RunJobFlow인수는 클러스터 구성에서 사용할 수 있습니다. 요청 구문에 대한 자세한 내용은 RunJobflow를 참조하세요.

다음 예시에서는 클러스터 구성을 EMR 단계 정의에 전달합니다. 그러면 새 EMR 클러스터에서 새 작업을 시작하는 단계가 표시됩니다. 이 예제의 EMR 클러스터 구성에는 기본 및 코어 EMR 클러스터 노드의 사양이 포함되어 있습니다. Amazon EMR 노드 유형에 대한 자세한 내용은 노드 유형 이해: 기본, 코어 및 태스크 노드를 참조하세요.

from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_step_config = EMRStepConfig(
    jar="jar-location", # required, path to jar file used
    args=["--verbose", "--force"], # optional list of arguments to pass to the jar
    main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest 
    properties=[ # optional list of Java properties that are set when the step runs
    {
        "key": "mapred.tasktracker.map.tasks.maximum",
        "value": "2"
    },
    {
        "key": "mapreduce.map.sort.spill.percent",
        "value": "0.90"
   },
   {
       "key": "mapreduce.tasktracker.reduce.tasks.maximum",
       "value": "5"
    }
  ]
)

# include your cluster configuration as a dictionary
emr_cluster_config = {
    "Applications": [
        {
            "Name": "Spark", 
        }
    ],
    "Instances":{
        "InstanceGroups":[
            {
                "InstanceRole": "MASTER",
                "InstanceCount": 1,
                "InstanceType": "m5.2xlarge"
            },
            {
                "InstanceRole": "CORE",
                "InstanceCount": 2,
                "InstanceType": "m5.2xlarge"
            }
        ]
    },
    "BootstrapActions":[],
    "ReleaseLabel": "emr-6.6.0",
    "JobFlowRole": "job-flow-role",
    "ServiceRole": "service-role"
}

emr_step = EMRStep(
    name="emr-step",
    cluster_id=None,
    display_name="emr_step",
    description="MyEMRStepDescription",
    step_config=emr_step_config,
    cluster_config=emr_cluster_config
)

전체 예시를 안내하는 샘플 노트북은 Pipelines EMR Step With Cluster Lifecycle Management를 참조하세요.

NotebookJobStep을 사용하여 SageMaker 노트북 작업을 파이프라인 단계로 비대화형으로 실행합니다. Pipelines 드래그 앤 드롭 UI에서 파이프라인을 빌드하는 경우 코드 실행 단계를 사용하여 노트북을 실행합니다. SageMaker 노트북 작업에 대한 자세한 정보는 SageMaker 노트북에서 확인하세요.

NotebookJobStep에는 최소한 입력 노트북, 이미지 URI 및 커널 이름이 필요합니다. 노트북 작업 단계 요구 사항 및 단계를 사용자 지정하기 위해 설정할 수 있는 기타 파라미터에 대한 자세한 내용은 sagemaker.workflow.steps.NotebookJobStep을 참조하세요.

다음 예시에서는 최소 인수를 사용하여 NotebookJobStep을 정의합니다.

from sagemaker.workflow.notebook_job_step import NotebookJobStep


notebook_job_step = NotebookJobStep(
    input_notebook=input_notebook,
    image_uri=image_uri,
    kernel_name=kernel_name
)

NotebookJobStep 파이프라인 단계는 SageMaker 노트북 작업으로 처리됩니다. 따라서 tags 인수에 특정 태그를 포함하여 Studio Classic UI 노트북 작업 대시보드에서 실행 상태를 추적합니다. 포함할 태그에 대한 자세한 내용은 Studio UI 대시보드에서 노트북 작업 보기 섹션을 참조하세요.

또한 SageMaker Python SDK를 사용하여 노트북 작업을 예약하는 경우 노트북 작업을 실행할 특정 이미지만 지정할 수 있습니다. 자세한 내용은 SageMaker AI Python SDK 노트북 작업에 대한 이미지 제약 조건 단원을 참조하십시오.

원하는 조건 또는 상태가 달성되지 않을 때 Amazon SageMaker Pipelines을 중지하는 데 실패 단계를 사용합니다. 또한 실패 단계를 사용하여 파이프라인 실행 실패의 원인을 나타내는 사용자 지정 오류 메시지를 입력할 수 있습니다.

실패 단계 사용에 대한 제한 사항

정적 텍스트 문자열 형태로 실패 단계 오류 메시지를 만들 수 있습니다. 또는 SDK를 사용할 경우 파이프라인 파라미터, 조인 작업 또는 기타 단계 속성을 사용하여 보다 유용한 오류 메시지를 만들 수도 있습니다.