기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.
Amazon SageMaker 모델 구축 파이프라인 단계
SageMaker 파이프라인은 단계로 구성됩니다. 이 단계는 파이프라인이 수행하는 작업과 속성을 사용하여 단계 간의 관계를 정의합니다.
단계 유형
다음은 각 단계 유형의 요구 사항을 설명하고 해당 단계의 예제 구현을 제공합니다. 이러한 구현은 필요한 리소스와 입력을 제공하지 않으므로 제대로 작동하지 않습니다. 이러한 단계를 구현하는 자습서는 SageMaker 파이프라인 생성 및 관리섹션을 참조하세요.
참고
데코레이터를 사용하여 로컬 머신러닝 코드를 SageMaker 파이프라인 단계로 변환하여 단계를 생성할 수도 있습니다. @step
자세한 내용은 @step 데코레이터 단원을 참조하십시오.
Amazon SageMaker 모델 구축 파이프라인은 다음 단계 유형을 지원합니다.
@step 데코레이터
@step
데코레이터를 사용하여 로컬 머신러닝 코드에서 단계를 만들 수 있습니다. 코드를 테스트한 후 데코레이터로 주석을 달아 함수를 SageMaker 파이프라인 단계로 변환할 수 있습니다. @step
SageMaker 파이프라인은 @step
-decated 함수의 출력을 파이프라인에 단계로 전달하면 파이프라인을 생성하고 실행합니다. 또한 기존 DAG 파이프라인 단계뿐 아니라 하나 이상의 @step
-데코레이티드 함수를 포함하는 다단계 파이프라인을 생성할 수도 있습니다. SageMaker @step
데코레이터로 단계를 만드는 방법에 대한 자세한 내용은 을 참조하십시오. L @step 데코레이터를 사용한 ift-and-shift Python 코드
처리 단계
처리 단계를 사용하여 데이터 처리를 위한 처리 작업을 생성합니다. 작업 처리에 대한 자세한 내용은 데이터 처리 및 모델 평가를 참조하세요.
처리 단계에는 프로세서, 처리 코드를 정의하는 Python 스크립트, 처리를 위한 출력 및 작업 인수가 필요합니다. 다음 예제는 ProcessingStep
정의를 생성하는 방법을 보여줍니다.
from sagemaker.sklearn.processing import SKLearnProcessor sklearn_processor = SKLearnProcessor(framework_version='1.0-1', role=
<role>
, instance_type='ml.m5.xlarge', instance_count=1)
from sagemaker.processing import ProcessingInput, ProcessingOutput from sagemaker.workflow.steps import ProcessingStep 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") ] step_process = ProcessingStep( name="AbaloneProcess", step_args = sklearn_processor.run(inputs=inputs, outputs=outputs, code="abalone/preprocessing.py") )
런타임 파라미터 전달
다음 예제는 PySpark 프로세서에서 a로 런타임 매개 변수를 전달하는 방법을 보여줍니다. ProcessingStep
from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.spark.processing import PySparkProcessor from sagemaker.processing import ProcessingInput, ProcessingOutput from sagemaker.workflow.steps import ProcessingStep pipeline_session = PipelineSession() pyspark_processor = PySparkProcessor( framework_version='2.4', role=
<role>
, instance_type='ml.m5.xlarge', instance_count=1, sagemaker_session=pipeline_session, ) step_args = pyspark_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="preprocess.py", arguments=None, ) step_process = ProcessingStep( name="AbaloneProcess", step_args=step_args, )
처리 단계 요구 사항에 대한 자세한 내용은 sagemaker.workflow.steps를 참조하십시오. ProcessingStep
훈련 단계
훈련 단계를 사용하여 모델을 훈련시키는 훈련 작업을 생성합니다. 교육 작업에 대한 자세한 내용은 Amazon을 통한 모델 학습을 참조하십시오 SageMaker.
훈련 단계에는 훈련 및 검증 데이터 입력뿐만 아니라 예측기가 필요합니다. 다음 예제는 TrainingStep
정의를 생성하는 방법을 보여줍니다. 교육 단계 요구 사항에 대한 자세한 내용은 sagemaker.workflow.steps를 참조하십시오. TrainingStep
from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.inputs import TrainingInput from sagemaker.workflow.steps import TrainingStep from sagemaker.xgboost.estimator import XGBoost pipeline_session = PipelineSession() xgb_estimator = XGBoost(..., sagemaker_session=pipeline_session) step_args = xgb_estimator.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="TrainAbaloneModel", step_args=step_args, )
튜닝 단계
조정 단계를 사용하여 하이퍼파라미터 최적화 () HPO 라고도 하는 하이퍼파라미터 조정 작업을 생성합니다. 하이퍼파라미터 조정 작업은 여러 교육 작업을 실행하며, 각 작업은 모델 버전을 생성합니다. 하이퍼파라미터 튜닝에 대한 자세한 내용은 를 사용하여 자동 모델 튜닝을 수행하십시오. SageMaker 섹션을 참조하세요.
튜닝 작업은 파이프라인의 SageMaker 실험과 연결되며, 교육 작업은 시험으로 생성됩니다. 자세한 내용은 실험 통합 단원을 참조하십시오.
조정 단계에는 AND 교육 입력이 HyperparameterTunerHyperparameterTuner
의 warm_start_config
파라미터를 지정하여 이전 튜닝 작업을 다시 훈련할 수 있습니다. 하이퍼파라미터 튜닝 및 부팅에 대한 자세한 내용은 웜 스타트 하이퍼파라미터 튜닝 작업 실행섹션을 참조하세요.
sagemaker.workflow.steps의 get_top_model_s3_uri
중요
조정 단계는 아마존 SageMaker 파이썬 SDK v2.48.0과 SageMaker 아마존 스튜디오 클래식 v3.8.0에서 도입되었습니다. 튜닝 단계를 사용하기 전에 스튜디오 클래식을 업데이트해야 합니다. 그렇지 않으면 파이프라인이 표시되지 않습니다. DAG 스튜디오 클래식을 업데이트하려면 을 참조하십시오 SageMaker 스튜디오 클래식 종료 및 업데이트.
다음 예제는 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://
my-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
AutoML 단계
AutoML을
참고
현재 AutoML 단계는 앙상블 훈련 모드만 지원합니다.
다음 예제는 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://my-bucket/my-training-data
", target_attribute_name="my_target_attribute_name
", channel_type="training", ) input_validation = AutoMLInput( inputs="s3://my-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
액세스하려면 an을 사용해야 합니다.
best_model = step_automl.get_best_auto_ml_model(
role=<role>
)
자세한 내용은 SageMaker Python의 AutoML
모델 단계
a를 ModelStep
사용하여 SageMaker 모델을 생성하거나 등록합니다. ModelStep
요구 사항에 대한 자세한 내용은 sagemaker.workflow.model_step을 참조하십시오. ModelStep
모델 생성
ModelStep
a를 사용하여 SageMaker 모델을 만들 수 있습니다. ModelStep
A에는 모델 아티팩트와 모델을 생성하는 데 사용해야 하는 SageMaker 인스턴스 유형에 대한 정보가 필요합니다. SageMaker 모델에 대한 자세한 내용은 Amazon을 통한 모델 학습을 참조하십시오 SageMaker.
다음 예제는 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"), )
모델 등록
ModelStep
a를 사용하여 Amazon SageMaker 모델 레지스트리에 a sagemaker.model.Model
또는 sagemaker.pipeline.PipelineModel
a를 등록할 수 있습니다. PipelineModel
은 추론 파이프라인은 데이터에 대해 추론 요청을 처리하는 컨테이너로 이루어진 선형 시퀀스로 구성된 모델입니다. 모델을 등록하는 자세한 방법은 모델 레지스트리에 모델 등록 및 배포섹션을 참조하세요.
다음 예제에서는 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, )
CreateModel 단계:
중요
Python v2.90.0부터 모델을 만들 때 사용하는 모델 단계 것이 좋습니다. SageMaker SDK CreateModelStep
이전 버전의 SageMaker Python에서 계속 SDK 작동하지만 더 이상 적극적으로 지원되지 않습니다.
CreateModel
단계를 사용하여 SageMaker 모델을 생성합니다. SageMaker 모델에 대한 자세한 내용은 Amazon을 통한 모델 학습을 참조하십시오 SageMaker.
모델 생성 단계에는 모델 생성에 사용해야 하는 모델 아티팩트와 SageMaker 인스턴스 유형에 대한 정보가 필요합니다. 다음 예제는 CreateModel
단계 정의를 생성하는 방법을 보여줍니다. CreateModel
단계 요구 사항에 대한 자세한 내용은 sagemaker.workflow.steps를 참조하십시오. CreateModelStep
from sagemaker.workflow.steps import CreateModelStep step_create_model = CreateModelStep( name="AbaloneCreateModel", model=best_model, inputs=inputs )
RegisterModel 단계
중요
Python v2.90.0부터 모델을 모델 단계 등록하는 데 사용하는 것이 좋습니다. SageMaker SDK RegisterModel
이전 버전의 SageMaker Python에서 계속 SDK 작동하지만 더 이상 적극적으로 지원되지 않습니다.
RegisterModel
단계를 사용하여 Sagemaker.Model.Model 또는 sagemaker.pipePipelineModel
은 추론 파이프라인은 데이터에 대해 추론 요청을 처리하는 컨테이너로 이루어진 선형 시퀀스로 구성된 모델입니다.
모델을 등록하는 자세한 방법은 모델 레지스트리에 모델 등록 및 배포섹션을 참조하세요. RegisterModel
단계 요구 사항에 대한 자세한 내용은 sagemaker.workflow.step_collections를 참조하십시오. RegisterModel
다음 예제에서는 PipelineModel
을 등록하는 RegisterModel
단계 생성 방법을 보여 줍니다.
import time from sagemaker.sklearn import SKLearnModel from sagemaker.xgboost import XGBoostModel 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=sagemaker_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=sagemaker_session, role=role) from sagemaker.workflow.step_collections import RegisterModel from sagemaker import PipelineModel pipeline_model = PipelineModel(models=[sklearn_model,xgboost_model],role=role,sagemaker_session=sagemaker_session) step_register = RegisterModel( name="AbaloneRegisterModel
", model=pipeline_model, 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
', )
model
이 제공되지 않은 경우 모델 등록 단계에는 다음 예제와 같이 예측기가 필요합니다.
from sagemaker.workflow.step_collections import RegisterModel 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 )
변환 단계
배치 변환을 위한 변환 단계를 사용하여 전체 데이터 세트에 대해 추론을 실행합니다. 배치 변환에 대한 자세한 정보는 추론 파이프라인을 사용하여 배치 변환 실행섹션을 참조하세요.
변환 단계에는 배치 변환을 실행하는 데 사용할 변환기와 데이터가 필요합니다. 다음 예제는 Transform
단계 정의를 생성하는 방법을 보여줍니다. Transform
단계 요구 사항에 대한 자세한 내용은 sagemaker.workflow.steps를 참조하십시오. TransformStep
from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.transformer import Transformer from sagemaker.inputs import TransformInput from sagemaker.workflow.steps import TransformStep transformer = Transformer(..., sagemaker_session=PipelineSession()) step_transform = TransformStep( name="AbaloneTransform", step_args=transformer.transform(data="s3://
my-bucket/my-data
"), )
조건 단계
조건 단계를 사용하여 단계 속성의 조건을 평가하여 파이프라인에서 다음에 취해야 할 조치를 평가합니다.
조건 단계에는 다음이 필요합니다.
-
조건 목록.
-
조건이 로
true
평가될 경우 실행할 단계 목록입니다. -
조건이 다음과 같이 평가될 경우 실행할 단계 목록입니다.
false
다음 예제는 ConditionStep
정의를 생성하는 방법을 보여줍니다.
제한 사항
-
SageMaker 파이프라인은 중첩된 조건 단계 사용을 지원하지 않습니다. 조건 단계를 다른 조건 단계의 입력으로 전달할 수 없습니다.
-
조건 단계는 두 브랜치에서 동일한 단계를 사용할 수 없습니다. 두 브랜치에 동일한 단계 기능이 필요한 경우 단계를 복제하고 다른 이름을 지정하세요.
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 ) step_cond = ConditionStep( name="AbaloneMSECond", conditions=[cond_lte], if_steps=[step_register, step_create_model, step_transform], else_steps=[] )
ConditionStep
요구 사항에 대한 자세한 내용은 sagemaker.workflow.condition_step을 참조하십시오. ConditionStep
콜백 단계
Callback
단계를 사용하여 Amazon SageMaker Model Building Pipeline에서 직접 제공하지 않는 추가 프로세스와 AWS 서비스를 워크플로에 추가하십시오. Callback
단계가 실행되면 다음 절차가 수행됩니다.
-
SageMaker 파이프라인은 고객이 지정한 Amazon 심플 큐 서비스 (SQSAmazon) 대기열로 메시지를 보냅니다. 메시지에는 SageMaker 파이프라인에서 생성한 토큰과 고객이 제공한 입력 파라미터 목록이 포함되어 있습니다. 메시지를 보낸 후 SageMaker 파이프라인은 고객의 응답을 기다립니다.
-
고객은 Amazon SQS 대기열에서 메시지를 검색하고 사용자 지정 프로세스를 시작합니다.
-
프로세스가 완료되면 고객은 다음 중 하나에 전화를 걸어 파이프라인에서 생성한 APIs 토큰을 제출합니다. SageMaker
-
SendPipelineExecutionStepSuccess, 출력 매개변수 목록과 함께
-
SendPipelineExecutionStepFailure, 실패 사유와 함께
-
-
이 API 호출로 인해 SageMaker Pipelines는 파이프라인 프로세스를 계속하거나 프로세스에 실패합니다.
Callback
단계 요구 사항에 대한 자세한 내용은 sagemaker.workflow.callback_step을 참조하십시오. CallbackStep
중요
Callback
단계는 아마존 SageMaker 파이썬 SDK v2.45.0과 SageMaker 아마존 스튜디오 클래식 v3.6.2에서 도입되었습니다. Callback
단계를 사용하기 전에 스튜디오 클래식을 업데이트해야 합니다. 그렇지 않으면 파이프라인이 표시되지 않습니다. DAG 스튜디오 클래식을 업데이트하려면 을 참조하십시오 SageMaker 스튜디오 클래식 종료 및 업데이트.
다음 샘플은 이전 절차의 구현을 보여줍니다.
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 to complete the step sagemaker_client.send_pipeline_execution_step_success( CallbackToken=token, OutputParameters={...} ) '
참고
CallbackStep
의 출력 파라미터는 중첩되지 않아야 합니다. 예를 들어, 중첩된 사전을 출력 파라미터로 사용하는 경우 사전은 단일 문자열로 취급됩니다 (예: {"output1": "{\"nested_output1\":\"my-output\"}"}
). 중첩된 값을 제공하는 경우 특정 출력 매개변수를 참조하려고 하면 재시도할 수 없는 SageMaker 클라이언트 오류가 발생합니다.
중지 동작
Callback
단계가 실행되는 동안에는 파이프라인 프로세스가 중지되지 않습니다.
실행 중인 Callback
단계가 있는 파이프라인 프로세스를 StopPipelineExecution호출하면 SageMaker Pipelines는 Amazon SQS 메시지를 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
에 호출을 추가합니다.
SageMaker 파이프라인은 이러한 호출 중 하나를 수신한 경우에만 파이프라인 프로세스를 중지합니다.
람다 스텝
Lambda 단계를 사용하여 함수를 실행합니다. AWS Lambda 기존 Lambda 함수를 SageMaker 실행하거나 새 Lambda 함수를 생성하고 실행할 수 있습니다. 파이프라인에서 Lambda 단계를 사용하는 방법을 보여주는 SageMaker 노트북은.ipynb를 참조하십시오. sagemaker-pipelines-lambda-step
중요
Lambda 단계는 아마존 SageMaker 파이썬 SDK v2.51.0과 아마존 스튜디오 클래식 v3.9.1에서 도입되었습니다. SageMaker Lambda 단계를 사용하기 전에 Studio Classic을 업데이트해야 합니다. 그렇지 않으면 DAG 파이프라인이 표시되지 않습니다. 스튜디오 클래식을 업데이트하려면 을 참조하십시오. SageMaker 스튜디오 클래식 종료 및 업데이트
SageMaker 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 ... )
세이지메이커.workflow.lambda_step. LambdaSteplambda_func
Lambda
기존 Lambda 함수를 호출하려면 함수의 Amazon 리소스 이름 ARN () 을 에 제공하기만 하면 됩니다. function_arn
function_arn
에 대한 값을 제공하지 않는 경우 handler
와 다음 중 하나를 지정해야 합니다.
-
zipped_code_dir
- 압축된 Lambda 함수의 경로s3_bucket
-zipped_code_dir
이 업로드될 Amazon S3 버킷 -
script
- 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_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
단계에서 정의해야 합니다.
참고
입력 및 출력 파라미터는 중첩되지 않아야 합니다. 예를 들어, 중첩된 사전을 출력 파라미터로 사용하는 경우 사전은 단일 문자열로 취급됩니다 (예: {"output1": "{\"nested_output1\":\"my-output\"}"}
).중첩된 값을 제공하고 나중에 참조하려고 하면 재시도할 수 없는 클라이언트 오류가 발생합니다.
Lambda
단계 정의 시, inputs
은 키-값 쌍의 딕셔너리여야 합니다. inputs
사전의 각 값은 프리미티브 유형(문자열, 정수 또는 부동)이어야 합니다. 중첩된 개체는 지원되지 않습니다. 정의되지 않은 상태로 두면 inputs
기본값은 None
입니다.
outputs
값은 키 목록이어야 합니다. 이 키는 Lambda
함수 출력에 정의된 사전을 참조합니다. inputs
과 마찬가지로 이러한 키는 프리미티브 유형이어야 하며 중첩된 객체는 지원되지 않습니다.
타임아웃 및 중지 동작
Lambda
클래스에는 Lambda 함수를 실행할 수 있는 최대 시간을 지정하는 timeout
인수가 있습니다. 기본값은 120초이며 최대값은 10분입니다. 제한 시간에 도달했을 때 Lambda 함수가 실행 중이면 Lambda 단계는 실패하지만 Lambda 함수는 계속 실행됩니다.
Lambda 단계에서 호출한 Lambda 함수를 중지할 수 없기 때문에 Lambda 단계가 실행되는 동안에는 파이프라인 프로세스를 중지할 수 없습니다. Lambda 함수가 실행되는 동안 프로세스를 중지하면 파이프라인은 함수가 완료될 때까지 또는 제한 시간에 도달할 때까지 기다립니다. 이는 어느 것이 먼저 발생하느냐에 따라 달라집니다. 그러면 프로세스가 중지됩니다. Lambda 함수가 완료되면 파이프라인 프로세스 상태는 Stopped
입니다. 제한 시간에 도달하면 파이프라인 프로세스 상태는 Failed
입니다.
ClarifyCheck 단계
이 ClarifyCheck
단계를 사용하여 바이어스 분석 및 모델 설명 가능성에 대해 이전 기준 대비 기준 드리프트 검사를 수행할 수 있습니다. 그런 다음 model.register()
메서드로 기준을 생성 및 등록하고 해당 메서드의 결과를 step_args
을 사용하여 모델 단계에 전달할 수 있습니다. Amazon SageMaker Model Monitor에서 모델 엔드포인트의 이러한 드리프트 점검 기준을 사용할 수 있습니다. 따라서 베이스라인 제안을 별도로 수행할 필요가 없습니다.
ClarifyCheck
단계에서는 모델 레지스트리에서 드리프트 검사의 기준을 가져올 수도 있습니다. 이 ClarifyCheck
단계에서는 사전 빌드된 Clarify SageMaker 컨테이너를 사용합니다. 이 컨테이너는 지정된 기준에 대한 제약조건 제안 및 제약조건 검증을 비롯한 다양한 모델 모니터링 기능을 제공합니다. 자세한 내용은 클래리파이 SageMaker 컨테이너로 시작하기 단원을 참조하십시오.
단계 ClarifyCheck 구성
파이프라인에서 사용할 때마다 사용 가능한 검사 유형 중 하나만을 수행하도록 ClarifyCheck
단계를 구성할 수 있습니다.
-
데이터 바이어스 검사
-
모델 바이어스 검사
-
모델 설명 가능성 검사
이렇게 하려면 다음 검사 유형 값 중 하나로 clarify_check_config
매개변수를 설정합니다.
-
DataBiasCheckConfig
-
ModelBiasCheckConfig
-
ModelExplainabilityCheckConfig
이 ClarifyCheck
단계에서는 사전 빌드된 Clarify SageMaker 컨테이너를 실행하는 처리 작업을 시작하며 검사 및 처리 작업을 위한 전용 구성이 필요합니다. ClarifyCheckConfig
및 CheckJobConfig
는 이러한 구성을 위한 도우미 함수입니다. 이러한 도우미 함수는 모델 편향, 데이터 편향 또는 모델 SageMaker 설명 가능성을 확인하기 위한 Clarify 처리 작업의 계산 방식과 일치합니다. 자세한 내용은 편향 분석 및 설명 가능성을 위한 처리 SageMaker 명확화 작업 실행 단원을 참조하십시오.
드리프트 검사의 단계 동작 제어
ClarifyCheck
단계의 동작을 제어하려면 다음 두 개의 부울 플래그가 필요합니다.
-
skip_check
: 이 파라미터는 이전 기준에 대한 드리프트 검사를 건너뛰었는지 여부를 나타냅니다.False
로 설정된 경우 구성된 검사 유형의 이전 기준을 사용할 수 있어야 합니다. -
register_new_baseline
: 이 파라미터는BaselineUsedForDriftCheckConstraints
단계 속성을 통해 새로 계산된 기준에 액세스할 수 있는지 여부를 나타냅니다.False
로 설정된 경우 구성된 검사 유형의 이전 기준 역시 사용할 수 있어야 합니다.BaselineUsedForDriftCheckConstraints
속성을 통해 액세스할 수 있습니다.
자세한 내용은 Amazon SageMaker 모델 구축 파이프라인의 기준 계산, 드리프트 감지 ClarifyCheck 및 수명 주기 및 QualityCheck 단계섹션을 참조하세요.
기준 작업
선택적으로 를 지정하여 기존 베이스라인을 찾을 수 있습니다model_package_group_name
. 그런 다음 이 ClarifyCheck
단계에서는 모델 패키지 그룹에서 승인된 최신 모델 패키지를 가져옵니다. DriftCheckBaselines
또는 supplied_baseline_constraints
파라미터를 통해 이전 기준을 제공할 수 있습니다. model_package_group_name
, supplied_baseline_constraints
를 모두 지정하는 경우 ClarifyCheck
단계에서는 supplied_baseline_constraints
파라미터로 지정된 기준을 사용합니다.
ClarifyCheck
단계 요구 사항 사용에 대한 자세한 내용은 sagemaker.workflow.steps를 참조하십시오. ClarifyCheck
단계를 사용하는 방법을 보여주는 Amazon SageMaker 스튜디오 클래식 노트북은 sagemaker-pipeline-model-monitor-clarify-steps.ipynb를
예 데이터 바이어스 검사를 위한 ClarifyCheck
단계 생성
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 단계:
이 QualityCheck
단계를 사용하여 파이프라인의 데이터 품질 또는 모델 품질에 대한 이전 기준선과 비교하여 베이스라인 제안 및 드리프트 검사를 수행합니다. 그런 다음 메서드를 사용하여 베이스라인을 생성 및 등록하고 해당 model.register()
메서드의 출력을 using에 전달할 수 있습니다.] 모델 단계 step_args
Model Monitor는 이러한 기준을 모델 엔드포인트의 드리프트 검사에 사용할 수 있으므로 기준 제안을 별도로 수행할 필요가 없습니다. QualityCheck
단계에서는 모델 레지스트리에서 드리프트 검사의 기준을 가져올 수도 있습니다. 이 QualityCheck
단계에서는 Amazon SageMaker Model Monitor의 사전 빌드된 컨테이너를 활용합니다. 이 컨테이너에는 제약조건 제안, 통계 생성, 기준선에 따른 제약조건 검증을 비롯한 다양한 모델 모니터링 기능이 있습니다. 자세한 내용은 Amazon SageMaker 모델 모니터 사전 제작 컨테이너 단원을 참조하십시오.
단계 QualityCheck 구성
파이프라인에서 사용할 때마다 다음 검사 유형 중 하나만 실행하도록 QualityCheck
단계를 구성할 수 있습니다.
-
데이터 품질 검사
-
모델 품질 검사
다음 검사 유형 값 중 하나로 quality_check_config
파라미터를 설정하여 이 작업을 수행합니다.
-
DataQualityCheckConfig
-
ModelQualityCheckConfig
QualityCheck
단계에서는 Model Monitor 사전 빌드된 컨테이너를 실행하는 처리 작업을 시작하며 검사 및 처리 작업을 위한 전용 구성이 필요합니다. QualityCheckConfig
및 CheckJobConfig
는 이러한 구성을 위한 도우미 함수입니다. 이러한 도우미 함수는 Model Monitor가 모델 품질 또는 데이터 품질 모니터링을 위한 기준을 생성하는 방식과 일치합니다. 모델 모니터 기준 제안에 대한 자세한 내용은 기준 생성, 모델 품질 베이스라인 생성섹션을 참조하세요.
드리프트 검사의 단계 동작 제어
QualityCheck
단계의 동작을 제어하려면 다음 두 개의 부울 플래그가 필요합니다.
-
skip_check
: 이 파라미터는 이전 기준에 대한 드리프트 검사를 건너뛰었는지 여부를 나타냅니다.False
로 설정된 경우 구성된 검사 유형의 이전 기준을 사용할 수 있어야 합니다. -
register_new_baseline
: 이 파라미터는 단계 속성BaselineUsedForDriftCheckConstraints
및BaselineUsedForDriftCheckStatistics
를 통해 새로 계산된 기준에 액세스할 수 있는지 여부를 나타냅니다.False
로 설정된 경우 구성된 검사 유형의 이전 기준 역시 사용할 수 있어야 합니다.BaselineUsedForDriftCheckConstraints
및BaselineUsedForDriftCheckStatistics
속성을 통해 이러한 속성에 액세스할 수 있습니다.
자세한 내용은 Amazon SageMaker 모델 구축 파이프라인의 기준 계산, 드리프트 감지 ClarifyCheck 및 수명 주기 및 QualityCheck 단계섹션을 참조하세요.
기준 작업
및 매개변수를 통해 이전 베이스라인을 직접 지정할 수 있습니다. supplied_baseline_statistics
supplied_baseline_constraints
모델 패키지 그룹에서 승인된 최신 모델 패키지를 DriftCheckBaselines
가져오는 QualityCheck
단계와 단계를 지정할 수도 있습니다. model_package_group_name
다음을 지정하는 경우 QualityCheck
단계는 해당 단계의 검사 유형에 supplied_baseline_statistics
따라 supplied_baseline_constraints
지정된 베이스라인을 사용합니다. QualityCheck
-
model_package_group_name
-
supplied_baseline_constraints
-
supplied_baseline_statistics
QualityCheck
단계 요구 사항 사용에 대한 자세한 내용은 sagemaker.workflow.steps를 참조하십시오. QualityCheck
단계를 사용하는 방법을 보여주는 Amazon SageMaker 스튜디오 클래식 노트북은 sagemaker-pipeline-model-monitor-clarify-steps.ipynb를
예 데이터 품질 검사를 위한 QualityCheck
단계 생성
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
" )
EMR단계:
Amazon SageMaker 모델 구축 파이프라인 EMR단계를 사용하여 다음을 수행하십시오.
-
실행 중인 Amazon EMR 클러스터에서 Amazon EMR 단계를 처리합니다.
-
파이프라인에서 Amazon EMR 클러스터를 생성하고 관리하도록 하십시오.
Amazon에 대한 자세한 내용은 Amazon EMR 시작하기를 참조하십시오EMR.
이 EMR 단계에는 Amazon EMR 클러스터에서 사용하는 JAR 파일의 위치와 전달될 인수를 EMRStepConfig
포함해야 합니다. 실행 중인 EMR 클러스터에서 단계를 실행하려는 경우에도 Amazon EMR 클러스터 ID를 제공하십시오. 또한 클러스터 구성을 전달하여 자동으로 생성, 관리 및 종료하는 클러스터에서 EMR 단계를 실행할 수 있습니다. 다음 섹션에는 두 가지 방법을 모두 보여주는 샘플 노트북으로 연결되는 링크와 예제가 포함되어 있습니다.
참고
-
EMR단계를 수행하려면 파이프라인에 전달된 역할에 추가 권한이 있어야 합니다. AWS 관리형 정책을 파이프라인 역할에 연결하거나 역할에 해당 정책의 권한이 포함되어 있는지 확인하세요.
AmazonSageMakerPipelinesIntegrations
-
EMREMR서버리스에서는 단계가 지원되지 않습니다. 또한 EMR Amazon에서는 지원되지 않습니다EKS.
-
실행 중인 클러스터에서 EMR 단계를 처리하는 경우 다음 상태 중 하나에 있는 클러스터만 사용할 수 있습니다.
-
STARTING
-
BOOTSTRAPPING
-
RUNNING
-
WAITING
-
-
실행 중인 클러스터에서 EMR 단계를 처리하는 경우 클러스터의 한
PENDING
상태에는 최대 256개의 EMR 단계가 있을 수 있습니다. EMR EMR이 한도를 초과하여 제출된 단계는 파이프라인 실행에 실패합니다. 파이프라인 단계에 대한 재시도 정책 사용을 고려해 볼 수 있습니다. 클러스터 ID와 클러스터 구성 중 하나를 지정할 수 있지만 둘 다 지정할 수는 없습니다.
이 EMR 단계에서는 Amazon에서 EMR 단계 또는 클러스터 상태의 변경 EventBridge 사항을 모니터링합니다. 실행 중인 클러스터에서 Amazon EMR 작업을 처리하는 경우 단계는
SageMakerPipelineExecutionEMRStepStatusUpdateRule
규칙을 사용하여 EMR EMR 단계 상태를 모니터링합니다. EMR단계가 생성하는 클러스터에서 작업을 처리하는 경우 단계는SageMakerPipelineExecutionEMRClusterStatusRule
규칙을 사용하여 클러스터 상태의 변경을 모니터링합니다. AWS 계정에 이러한 EventBridge 규칙 중 하나가 표시되면 삭제하지 마십시오. 그렇지 않으면 EMR 단계가 완료되지 않을 수 있습니다.
실행 중인 Amazon EMR 클러스터에서 새 작업 시작
실행 중인 Amazon EMR 클러스터에서 새 작업을 시작하려면 클러스터 ID를 의 cluster_id
인수에 문자열로 EMRStep
전달하십시오. 다음 예에서는 이 절차를 보여 줍니다.
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" )
전체 예제를 안내하는 샘플 노트북은 EMR클러스터 실행을 포함한 SageMaker 파이프라인 EMR 단계를
새 Amazon EMR 클러스터에서 새 작업 시작
자동으로 EMRStep
생성되는 새 클러스터에서 새 작업을 시작하려면 클러스터 구성을 사전으로 제공하십시오. 사전은 RunJobFlow요청과 구조가 같아야 합니다. 하지만 클러스터 구성에 다음 필드를 포함시키지 마세요.
[
Name
][
Steps
][
AutoTerminationPolicy
][
Instances
][KeepJobFlowAliveWhenNoSteps
][
Instances
][TerminationProtected
]
다른 모든 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 )
전체 예제를 안내하는 샘플 노트북은 클러스터 수명 주기 관리를 SageMaker 포함한 파이프라인 EMR 단계를
노트북 작업 단계
NotebookJobStep
를 사용하여 SageMaker Notebook Job을 비대화형 방식으로 파이프라인 단계로 실행할 수 있습니다. SageMaker Notebook 작업에 대한 자세한 내용은 을 참조하십시오. SageMaker 노트북 작업
NotebookJobStep
A에는 최소한 입력 노트북, 이미지 URI 및 커널 이름이 필요합니다. Notebook Job 단계 요구 사항 및 단계를 사용자 정의하기 위해 설정할 수 있는 기타 매개변수에 대한 자세한 내용은 sagemaker.workflow.steps를 참조하십시오. NotebookJobStep
다음 예제에서는 최소 인수를 사용하여 a를 정의합니다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 SDK Python을 사용하여 노트북 작업을 예약하는 경우 특정 이미지만 지정하여 노트북 작업을 실행할 수 있습니다. 자세한 내용은 SageMakerPython SDK 노트북 작업에 대한 이미지 제약 단원을 참조하십시오.
실패 단계
원하는 조건이나 상태가 달성되지 않을 때 FailStep
a를 사용하여 Amazon SageMaker Model Building 파이프라인 실행을 중지할 수 있습니다. 또한 이 경우 해당 파이프라인의 실행이 실패로 표시됩니다. FailStep
은 또한 파이프라인 실행 실패의 원인을 나타내는 사용자 지정 오류 메시지를 입력할 수 있습니다.
참고
파이프라인 FailStep
단계와 다른 파이프라인 단계가 동시에 실행되면 모든 동시 단계가 완료될 때까지 파이프라인이 종료되지 않습니다.
FailStep
사용 제한 사항
-
다른 단계
DependsOn
목록에FailStep
을 추가할 수 없습니다. 자세한 내용은 단계 간 사용자 지정 종속성섹션을 참조하세요. -
다른 단계는
FailStep
을 참조할 수 없습니다. 이는 항상 파이프라인 실행의 마지막 단계입니다. -
FailStep
으로 끝나는 파이프라인 실행은 재시도할 수 없습니다.
정적 텍스트 문자열 형태로 FailStep
ErrorMessage
를 생성할 수 있습니다. 또는 파이프라인 파라미터, 조인
다음 예제 코드 스니펫은 파이프라인 파라미터로 구성된 ErrorMessage
와 Join
작업을 사용하는 FailStep
을 사용합니다.
from sagemaker.workflow.fail_step import FailStep from sagemaker.workflow.functions import Join from sagemaker.workflow.parameters import ParameterInteger mse_threshold_param = ParameterInteger(name="MseThreshold", default_value=5) step_fail = FailStep( name="
AbaloneMSEFail
", error_message=Join( on=" ", values=["Execution failed due to MSE >
", mse_threshold_param] ), )
단계 속성
properties
속성을 사용하여 파이프라인의 단계 간에 데이터 종속성을 추가할 수 있습니다. SageMaker 파이프라인은 이러한 데이터 종속성을 사용하여 파이프라인 DAG 정의로부터 데이터를 생성합니다. 이러한 속성은 자리 표시자 값으로 참조될 수 있으며 런타임에 확인됩니다.
SageMaker 파이프라인 단계의 properties
속성은 해당 SageMaker 작업 유형에 대한 Describe
호출에서 반환된 객체와 일치합니다. 각 작업 유형에 대해 Describe
호출은 다음과 같은 응답 객체를 반환합니다.
-
ProcessingStep
– DescribeProcessingJob -
TrainingStep
– DescribeTrainingJob -
TransformStep
– DescribeTransformJob
데이터 종속성 생성 중에 각 단계 유형에 대해 참조할 수 있는 속성을 확인하려면 Amazon Python의 데이터 종속성 - 속성
단계 병렬성
단계가 다른 단계에 종속되지 않는 경우 파이프라인 실행 즉시 실행됩니다. 하지만 너무 많은 파이프라인 단계를 병렬로 실행하면 가용 리소스가 빠르게 고갈될 수 있습니다. ParallelismConfiguration
을 사용하여 파이프라인 실행의 동시 단계 수를 제어할 수 있습니다.
다음 예제에서는 ParallelismConfiguration
을 사용하여 동시 단계 제한을 5로 설정합니다.
pipeline.create( parallelism_config=ParallelismConfiguration(5), )
단계 간 데이터 종속성
단계 간의 데이터 관계를 DAG 지정하여 구조를 정의합니다. 단계 간에 데이터 종속성을 만들려면 한 단계의 속성을 파이프라인의 다른 단계에 입력으로 전달합니다. 입력을 받는 단계는 입력을 제공하는 단계가 실행을 완료할 때까지 시작되지 않습니다.
데이터 종속성은 다음 형식의 JsonPath 표기법을 사용합니다. 이 형식은 속성 파일을 탐색합니다. JSON 즉, 여러 개를 추가할 수 있습니다.<property>
파일에서 원하는 중첩 속성에 도달하는 데 필요한 만큼의 인스턴스 JsonPath 표기법에 대한 자세한 내용은 리포지토리를 참조하십시오. JsonPath
<step_name>
.properties.<property>
.<property>
다음은 처리 단계의 ProcessingOutputConfig
속성을 사용하여 Amazon S3 버킷을 지정하는 방법을 보여줍니다.
step_process.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri
데이터 종속성을 생성하려면 다음과 같이 버킷을 훈련 단계로 전달하세요.
from sagemaker.workflow.pipeline_context import PipelineSession sklearn_train = SKLearn(..., sagemaker_session=PipelineSession()) step_train = TrainingStep( name="CensusTrain", step_args=sklearn_train.fit(inputs=TrainingInput( s3_data=step_process.properties.ProcessingOutputConfig.Outputs[ "train_data"].S3Output.S3Uri )) )
데이터 종속성 생성 중에 각 단계 유형에 대해 참조할 수 있는 속성을 확인하려면 Amazon Python의 데이터 종속성 - 속성
단계 간 사용자 지정 종속성
데이터 종속성을 지정하면 SageMaker 파이프라인은 단계 간 데이터 연결을 제공합니다. 또는 파이프라인을 직접 사용하지 SageMaker 않고도 한 단계에서 이전 단계의 데이터에 액세스할 수 있습니다. 이 경우 다른 단계의 실행이 완료될 때까지 단계를 시작하지 않도록 SageMaker 파이프라인에 지시하는 사용자 지정 종속성을 생성할 수 있습니다. 단계의 DependsOn
속성을 지정하여 사용자 지정 종속성을 생성합니다.
예를 들어, 다음은 A
단계와 B
단계가 모두 실행된 후에만 시작되는 C
단계를 정의합니다.
{ 'Steps': [ {'Name':'A', ...}, {'Name':'B', ...}, {'Name':'C', 'DependsOn': ['A', 'B']} ] }
SageMaker 종속 항목으로 인해 순환 종속성이 생성되는 경우 파이프라인에서 유효성 검사 예외가 발생합니다.
다음 예제는 처리 단계 실행이 끝난 후 시작되는 훈련 단계를 생성합니다.
processing_step = ProcessingStep(...) training_step = TrainingStep(...) training_step.add_depends_on([processing_step])
다음 예제에서는 서로 다른 두 처리 단계의 실행이 완료될 때까지 시작되지 않는 훈련 단계를 만듭니다.
processing_step_1 = ProcessingStep(...) processing_step_2 = ProcessingStep(...) training_step = TrainingStep(...) training_step.add_depends_on([processing_step_1, processing_step_2])
다음은 사용자 지정 종속성을 생성하는 다른 방법을 제공합니다.
training_step.add_depends_on([processing_step_1]) training_step.add_depends_on([processing_step_2])
다음 예제에서는 한 처리 단계에서 입력을 받고 다른 처리 단계가 실행을 마칠 때까지 기다리는 훈련 단계를 만듭니다.
processing_step_1 = ProcessingStep(...) processing_step_2 = ProcessingStep(...) training_step = TrainingStep( ..., inputs=TrainingInput( s3_data=processing_step_1.properties.ProcessingOutputConfig.Outputs[ "train_data" ].S3Output.S3Uri ) training_step.add_depends_on([processing_step_2])
다음 예제에서는 단계의 사용자 정의 종속성에 대한 문자열 목록을 검색하는 방법을 보여줍니다.
custom_dependencies = training_step.depends_on
한 단계에서 사용자 지정 이미지 사용
파이프라인에서 단계를 생성할 때 사용 가능한 모든 SageMaker 딥러닝 컨테이너 이미지를
파이프라인 단계가 있는 자체 컨테이너를 사용할 수도 있습니다. Studio Classic 내에서 이미지를 만들 수 없으므로 SageMaker 파이프라인에서 이미지를 사용하기 전에 다른 방법을 사용하여 이미지를 만들어야 합니다.
파이프라인 단계를 만들 때 자체 컨테이너를 사용하려면 추정기 URI 정의에 이미지를 포함하세요. 에서 자체 컨테이너를 사용하는 방법에 대한 자세한 내용은 Docker 컨테이너 사용을 참조하십시오. SageMaker SageMaker