Langkah-langkah Pipa Pembuatan SageMaker Model Amazon

SageMaker Pipa terdiri dari langkah-langkah. Langkah-langkah ini menentukan tindakan yang diambil pipeline dan hubungan antara langkah-langkah menggunakan properti.

Jenis langkah

Berikut ini menjelaskan persyaratan dari setiap jenis langkah dan memberikan contoh implementasi langkah. Ini bukan implementasi yang berfungsi karena mereka tidak menyediakan sumber daya dan input yang dibutuhkan. Untuk tutorial yang mengimplementasikan langkah-langkah ini, lihatMembuat dan Mengelola SageMaker Pipelines.

Amazon SageMaker Model Building Pipelines mendukung jenis langkah berikut:

@step dekorator

Anda dapat membuat langkah dari kode pembelajaran mesin lokal menggunakan @step dekorator. Setelah Anda menguji kode Anda, Anda dapat mengonversi fungsi ke langkah SageMaker pipeline dengan membubuhi keterangan dengan dekorator. @step SageMaker Pipelines membuat dan menjalankan pipa ketika Anda melewati output dari fungsi @step -decorated sebagai langkah ke pipeline Anda. Anda juga dapat membuat DAG pipa multi-langkah yang mencakup satu atau lebih fungsi @step yang didekorasi serta langkah-langkah SageMaker pipa tradisional. Untuk detail selengkapnya tentang cara membuat langkah dengan @step dekorator, lihatL Kode ift-and-shift Python dengan dekorator @step.

Langkah pemrosesan

Gunakan langkah pemrosesan untuk membuat pekerjaan pemrosesan untuk pemrosesan data. Untuk informasi selengkapnya tentang pekerjaan pemrosesan, lihat Memproses Data dan Mengevaluasi Model.

Langkah pemrosesan membutuhkan prosesor, skrip Python yang mendefinisikan kode pemrosesan, output untuk pemrosesan, dan argumen pekerjaan. Contoh berikut menunjukkan cara membuat ProcessingStep definisi.

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

Lulus parameter runtime

Contoh berikut menunjukkan cara meneruskan parameter runtime dari PySpark prosesor ke file. 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,
)

Untuk informasi selengkapnya tentang persyaratan langkah pemrosesan, lihat sagemaker.workflow.steps. ProcessingStepdokumentasi. Untuk contoh mendalam, lihat buku catatan contoh Orchestrate Jobs to Train and Evaluate Models with Amazon SageMaker Pipelines. Bagian Tentukan Langkah Pemrosesan untuk Rekayasa Fitur mencakup informasi lebih lanjut.

Langkah pelatihan

Anda menggunakan langkah pelatihan untuk membuat pekerjaan pelatihan untuk melatih model. Untuk informasi selengkapnya tentang pekerjaan pelatihan, lihat Melatih Model dengan Amazon SageMaker.

Langkah pelatihan membutuhkan estimator, serta input data pelatihan dan validasi. Contoh berikut menunjukkan cara membuat TrainingStep definisi. Untuk informasi selengkapnya tentang persyaratan langkah pelatihan, lihat sagemaker.workflow.steps. TrainingStepdokumentasi.

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

Langkah penyetelan

Anda menggunakan langkah penyetelan untuk membuat pekerjaan tuning hyperparameter, juga dikenal sebagai optimasi hyperparameter (). HPO Pekerjaan tuning hyperparameter menjalankan beberapa pekerjaan pelatihan, dengan setiap pekerjaan menghasilkan versi model. Untuk informasi lebih lanjut tentang penyetelan hyperparameter, lihat. Lakukan penyetelan model otomatis dengan SageMaker

Pekerjaan penyetelan dikaitkan dengan SageMaker eksperimen untuk pipa, dengan pekerjaan pelatihan yang dibuat sebagai uji coba. Untuk informasi selengkapnya, lihat Integrasi Eksperimen.

Langkah penyetelan membutuhkan input HyperparameterTunerdan pelatihan. Anda dapat melatih kembali pekerjaan penyetelan sebelumnya dengan menentukan warm_start_config parameter. HyperparameterTuner Untuk informasi lebih lanjut tentang penyetelan hyperparameter dan start hangat, lihat. Jalankan Pekerjaan Tuning Hyperparameter Mulai yang Hangat

Anda menggunakan metode get_top_model_s3_uri dari sagemaker.workflow.steps. TuningStepkelas untuk mendapatkan artefak model dari salah satu versi model berkinerja terbaik. Untuk buku catatan yang menunjukkan cara menggunakan langkah penyetelan dalam SageMaker pipeline, lihat sagemaker-pipelines-tuning-step.ipynb.

Contoh berikut menunjukkan cara membuat TuningStep definisi.

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

Dapatkan versi model terbaik

Contoh berikut menunjukkan cara mendapatkan versi model terbaik dari pekerjaan penyetelan menggunakan get_top_model_s3_uri metode ini. Paling-paling, 50 versi berkinerja teratas tersedia berdasarkan peringkat HyperParameterTuningJobObjective. top_kArgumennya adalah indeks ke dalam versi, di mana top_k=0 versi berkinerja terbaik dan top_k=49 merupakan versi berkinerja terburuk.

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

Untuk informasi selengkapnya tentang persyaratan langkah penyetelan, lihat sagemaker.workflow.steps. TuningStepdokumentasi.

Langkah AutoML

Gunakan AutoML API untuk membuat pekerjaan AutoML untuk melatih model secara otomatis. Untuk informasi selengkapnya tentang pekerjaan AutoML, lihat Mengotomatiskan pengembangan model dengan Amazon Autopilot. SageMaker

Contoh berikut menunjukkan cara membuat definisi menggunakanAutoMLStep.

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

Dapatkan versi model terbaik

Langkah AutoML secara otomatis melatih beberapa kandidat model. Dapatkan model dengan metrik objektif terbaik dari pekerjaan AutoML menggunakan get_best_auto_ml_model metode sebagai berikut. Anda juga harus menggunakan artefak model IAM role untuk mengakses.

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

Untuk informasi selengkapnya, lihat langkah AutoML di Python SageMaker . SDK

Langkah model

Gunakan a ModelStep untuk membuat atau mendaftarkan SageMaker model. Untuk informasi selengkapnya tentang ModelStep persyaratan, lihat sagemaker.workflow.model_step. ModelStepdokumentasi.

Buat model

Anda dapat menggunakan a ModelStep untuk membuat SageMaker model. A ModelStep membutuhkan artefak model dan informasi tentang jenis SageMaker instance yang perlu Anda gunakan untuk membuat model. Untuk informasi selengkapnya tentang SageMaker model, lihat Melatih Model dengan Amazon SageMaker.

Contoh berikut menunjukkan cara membuat ModelStep definisi.

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

Daftarkan model

Anda dapat menggunakan a ModelStep untuk mendaftarkan a sagemaker.model.Model atau sagemaker.pipeline.PipelineModel dengan registri SageMaker model Amazon. A PipelineModel mewakili pipa inferensi, yang merupakan model yang terdiri dari urutan linier kontainer yang memproses permintaan inferensi. Untuk informasi selengkapnya tentang cara mendaftarkan model, lihatDaftarkan dan Terapkan Model dengan Registri Model.

Contoh berikut menunjukkan cara membuat ModelStep yang mendaftarkan aPipelineModel.

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 langkah

Anda menggunakan CreateModel langkah untuk membuat SageMaker model. Untuk informasi selengkapnya tentang SageMaker model, lihat Melatih Model dengan Amazon SageMaker.

Langkah membuat model memerlukan artefak model dan informasi tentang jenis SageMaker instance yang perlu Anda gunakan untuk membuat model. Contoh berikut menunjukkan cara membuat definisi CreateModel langkah. Untuk informasi selengkapnya tentang persyaratan CreateModel langkah, lihat sagemaker.workflow.steps. CreateModelStepdokumentasi.

from sagemaker.workflow.steps import CreateModelStep

step_create_model = CreateModelStep(
    name="AbaloneCreateModel",
    model=best_model,
    inputs=inputs
)

RegisterModel langkah

Anda menggunakan RegisterModel langkah untuk mendaftarkan Sagemaker.model.Model atau sagemaker.pipeline. PipelineModeldengan registri SageMaker model Amazon. A PipelineModel mewakili pipa inferensi, yang merupakan model yang terdiri dari urutan linier kontainer yang memproses permintaan inferensi.

Untuk informasi selengkapnya tentang cara mendaftarkan model, lihatDaftarkan dan Terapkan Model dengan Registri Model. Untuk informasi selengkapnya tentang persyaratan RegisterModel langkah, lihat sagemaker.workflow.step_collections. RegisterModeldokumentasi.

Contoh berikut menunjukkan cara membuat RegisterModel langkah yang mendaftarkan aPipelineModel.

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

Jika model tidak disediakan, langkah model register memerlukan estimator seperti yang ditunjukkan pada contoh berikut.

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
)

Transformasikan langkah

Anda menggunakan langkah transformasi untuk transformasi batch untuk menjalankan inferensi pada seluruh kumpulan data. Untuk informasi selengkapnya tentang transformasi batch, lihatJalankan Transformasi Batch dengan Pipa Inferensi.

Langkah transformasi membutuhkan transformator dan data untuk menjalankan transformasi batch. Contoh berikut menunjukkan cara membuat definisi Transform langkah. Untuk informasi selengkapnya tentang persyaratan Transform langkah, lihat sagemaker.workflow.steps. TransformStepdokumentasi.

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

Langkah kondisi

Anda menggunakan langkah kondisi untuk mengevaluasi kondisi properti langkah untuk menilai tindakan mana yang harus diambil selanjutnya dalam pipeline.

Langkah kondisi membutuhkan:

Contoh berikut menunjukkan cara membuat ConditionStep definisi.

Batasan

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=[]
)

Untuk informasi selengkapnya tentang ConditionStep persyaratan, lihat sagemaker.workflow.condition_step. ConditionStepAPIreferensi. Untuk informasi selengkapnya tentang kondisi yang didukung, lihat Pipa Pembuatan SageMaker Model Amazon - Ketentuan dalam dokumentasi SageMaker SDK Python.

Langkah panggilan balik

Gunakan Callback langkah untuk menambahkan proses dan AWS layanan tambahan ke dalam alur kerja Anda yang tidak langsung disediakan oleh Amazon SageMaker Model Building Pipelines. Ketika sebuah Callback langkah berjalan, prosedur berikut terjadi:

Untuk informasi selengkapnya tentang persyaratan Callback langkah, lihat sagemaker.workflow.callback_step. CallbackStepdokumentasi. Untuk solusi selengkapnya, lihat Memperluas SageMaker Pipelines untuk menyertakan langkah-langkah kustom menggunakan langkah panggilan balik.

Sampel berikut menunjukkan implementasi dari prosedur sebelumnya.

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={...}
            )
'

Menghentikan perilaku

Proses pipeline tidak berhenti saat Callback langkah sedang berjalan.

Saat Anda memanggil StopPipelineExecutionproses pipeline dengan Callback langkah yang sedang berjalan, SageMaker Pipelines mengirimkan SQS pesan Amazon ke SQS antrian. Isi SQS pesan berisi bidang Status, yang diatur keStopping. Berikut ini menunjukkan contoh badan SQS pesan.

{
  "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"
}

Anda harus menambahkan logika ke konsumen SQS pesan Amazon Anda untuk mengambil tindakan yang diperlukan (misalnya, pembersihan sumber daya) setelah menerima pesan. Kemudian tambahkan panggilan ke SendPipelineExecutionStepSuccess atauSendPipelineExecutionStepFailure.

Hanya ketika SageMaker Pipelines menerima salah satu panggilan ini barulah ia menghentikan proses pipeline.

Langkah Lambda

Anda menggunakan langkah Lambda untuk menjalankan fungsi. AWS Lambda Anda dapat menjalankan fungsi Lambda yang ada, atau SageMaker dapat membuat dan menjalankan fungsi Lambda baru. Untuk buku catatan yang menunjukkan cara menggunakan langkah Lambda dalam SageMaker pipeline, lihat sagemaker-pipelines-lambda-step .ipynb.

SageMaker menyediakan kelas SageMaker.lambda_helper.Lambda untuk membuat, memperbarui, memanggil, dan menghapus fungsi Lambda. Lambdamemiliki tanda tangan berikut.

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. LambdaStepkelas memiliki lambda_func argumen tipeLambda. Untuk menjalankan fungsi Lambda yang ada, satu-satunya persyaratan adalah menyediakan Amazon Resource Name ARN () dari fungsi tersebut. function_arn Jika Anda tidak memberikan nilai untukfunction_arn, Anda harus menentukan handler dan salah satu dari berikut ini:

Contoh berikut menunjukkan cara membuat definisi Lambda langkah yang memanggil fungsi Lambda yang ada.

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

Contoh berikut menunjukkan cara membuat definisi Lambda langkah yang membuat dan memanggil fungsi Lambda menggunakan skrip fungsi 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"
    ]
)

Input dan output

Jika Lambda fungsi Anda memiliki input atau output, ini juga harus didefinisikan dalam langkah AndaLambda.

Saat mendefinisikan Lambda langkah, inputs harus kamus pasangan kunci-nilai. Setiap nilai inputs kamus harus berupa tipe primitif (string, integer, atau float). Objek bersarang tidak didukung. Jika dibiarkan tidak terdefinisi, inputs nilai defaultnya. None

outputsNilai harus berupa daftar kunci. Kunci-kunci ini mengacu pada kamus yang didefinisikan dalam output Lambda fungsi. Sepertiinputs, kunci ini harus tipe primitif, dan objek bersarang tidak didukung.

Timeout dan menghentikan perilaku

LambdaKelas memiliki timeout argumen yang menentukan waktu maksimum bahwa fungsi Lambda dapat berjalan. Nilai default adalah 120 detik dengan nilai maksimum 10 menit. Jika fungsi Lambda berjalan saat batas waktu terpenuhi, langkah Lambda gagal; Namun, fungsi Lambda terus berjalan.

Proses pipeline tidak dapat dihentikan saat langkah Lambda berjalan karena fungsi Lambda yang dipanggil oleh langkah Lambda tidak dapat dihentikan. Jika Anda menghentikan proses saat fungsi Lambda sedang berjalan, pipeline menunggu fungsi selesai atau sampai batas waktu tercapai. Ini tergantung pada mana yang terjadi lebih dulu. Prosesnya kemudian berhenti. Jika fungsi Lambda selesai, status proses pipeline adalah. Stopped Jika batas waktu tercapai, status proses pipeline adalahFailed.

ClarifyCheck langkah

Anda dapat menggunakan ClarifyCheck langkah untuk melakukan pemeriksaan penyimpangan dasar terhadap garis dasar sebelumnya untuk analisis bias dan penjelasan model. Anda kemudian dapat membuat dan mendaftarkan baseline Anda dengan model.register() metode dan meneruskan output dari metode itu untuk Langkah model digunakan. step_args Garis dasar untuk pemeriksaan drift ini dapat digunakan oleh Amazon SageMaker Model Monitor untuk titik akhir model Anda. Akibatnya, Anda tidak perlu melakukan saran dasar secara terpisah.

ClarifyCheckLangkah ini juga dapat menarik garis dasar untuk pemeriksaan drift dari registri model. ClarifyCheckLangkah ini menggunakan Container Prebuilt SageMaker Clarify. Wadah ini menyediakan berbagai kemampuan pemantauan model, termasuk saran kendala dan validasi kendala terhadap garis dasar yang diberikan. Untuk informasi selengkapnya, lihat Memulai dengan SageMaker Clarify Container.

Mengkonfigurasi langkahnya ClarifyCheck

Anda dapat mengonfigurasi ClarifyCheck langkah untuk melakukan hanya satu dari jenis pemeriksaan berikut setiap kali digunakan dalam pipeline.

Untuk melakukan ini, atur clarify_check_config parameter dengan salah satu nilai jenis cek berikut:

ClarifyCheckLangkah ini meluncurkan pekerjaan pemrosesan yang menjalankan kontainer bawaan SageMaker Clarify dan memerlukan konfigurasi khusus untuk pemeriksaan dan pekerjaan pemrosesan. ClarifyCheckConfigdan CheckJobConfig merupakan fungsi pembantu untuk konfigurasi ini. Fungsi pembantu ini selaras dengan bagaimana pekerjaan pemrosesan SageMaker Clarify menghitung untuk memeriksa bias model, bias data, atau penjelasan model. Untuk informasi selengkapnya, lihat Jalankan Pekerjaan Pemrosesan SageMaker Klarifikasi untuk Analisis Bias dan Penjelasan.

Mengontrol perilaku langkah untuk pemeriksaan drift

ClarifyCheckLangkah ini membutuhkan dua flag boolean berikut untuk mengontrol perilakunya:

Untuk informasi selengkapnya, lihat Perhitungan dasar, deteksi drift, dan siklus hidup dengan serta ClarifyCheck langkah-langkah QualityCheck di Amazon Model Building Pipelines SageMaker .

Bekerja dengan baseline

Anda dapat secara opsional menentukan model_package_group_name untuk menemukan baseline yang ada. Kemudian, ClarifyCheck langkah tersebut DriftCheckBaselines menarik paket model terbaru yang disetujui dalam grup paket model.

Atau, Anda dapat memberikan garis dasar sebelumnya melalui parameter. supplied_baseline_constraints Jika Anda menentukan kedua model_package_group_name dansupplied_baseline_constraints, ClarifyCheck langkah menggunakan garis dasar yang ditentukan oleh parameter. supplied_baseline_constraints

Untuk informasi selengkapnya tentang penggunaan persyaratan ClarifyCheck langkah, lihat sagemaker.workflow.steps. ClarifyCheckStepdi Amazon SageMaker SageMaker SDK untuk Python. Untuk notebook Amazon SageMaker Studio Classic yang menunjukkan cara menggunakan ClarifyCheck step di SageMaker Pipelines, lihat 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 langkah

Gunakan QualityCheck langkah untuk melakukan saran dasar dan pemeriksaan drift terhadap baseline sebelumnya untuk kualitas data atau kualitas model dalam pipeline. Anda kemudian dapat membuat dan mendaftarkan baseline Anda dengan model.register() metode dan meneruskan output dari metode itu untuk Langkah model digunakanstep_args.]

Model Monitor dapat menggunakan garis dasar ini untuk pemeriksaan drift untuk titik akhir model Anda sehingga Anda tidak perlu melakukan saran dasar secara terpisah. QualityCheckLangkah ini juga dapat menarik garis dasar untuk pemeriksaan drift dari registri model. QualityCheckLangkah ini memanfaatkan wadah prebuilt Amazon SageMaker Model Monitor. Wadah ini memiliki berbagai kemampuan pemantauan model termasuk saran kendala, pembuatan statistik, dan validasi kendala terhadap baseline. Untuk informasi selengkapnya, lihat Amazon SageMaker Model Monitor wadah bawaan.

Mengkonfigurasi langkahnya QualityCheck

Anda dapat mengonfigurasi QualityCheck langkah untuk menjalankan hanya satu dari jenis pemeriksaan berikut setiap kali digunakan dalam pipeline.

Anda melakukan ini dengan mengatur quality_check_config parameter dengan salah satu nilai jenis cek berikut:

QualityCheckLangkah ini meluncurkan pekerjaan pemrosesan yang menjalankan wadah bawaan Model Monitor dan memerlukan konfigurasi khusus untuk pemeriksaan dan pekerjaan pemrosesan. Fungsi QualityCheckConfig dan CheckJobConfig pembantu untuk konfigurasi ini. Fungsi pembantu ini selaras dengan bagaimana Model Monitor membuat garis dasar untuk kualitas model atau pemantauan kualitas data. Untuk informasi selengkapnya tentang saran dasar Model Monitor, lihat Buat Baseline dan. Buat baseline kualitas model

Mengontrol perilaku langkah untuk pemeriksaan drift

QualityCheckLangkah ini membutuhkan dua flag Boolean berikut untuk mengontrol perilakunya:

Untuk informasi selengkapnya, lihat Perhitungan dasar, deteksi drift, dan siklus hidup dengan serta ClarifyCheck langkah-langkah QualityCheck di Amazon Model Building Pipelines SageMaker .

Bekerja dengan baseline

Anda dapat menentukan garis dasar sebelumnya secara langsung melalui parameter supplied_baseline_statistics dansupplied_baseline_constraints. Anda juga dapat menentukan model_package_group_name dan QualityCheck langkah menarik paket model terbaru yang disetujui dalam grup paket model. DriftCheckBaselines

Saat Anda menentukan yang berikut ini, QualityCheck langkah menggunakan garis dasar yang ditentukan oleh supplied_baseline_constraints dan supplied_baseline_statistics pada jenis pemeriksaan langkah. QualityCheck

Untuk informasi selengkapnya tentang penggunaan persyaratan QualityCheck langkah, lihat sagemaker.workflow.steps. QualityCheckStepdi Amazon SageMaker SageMaker SDK untuk Python. Untuk notebook Amazon SageMaker Studio Classic yang menunjukkan cara menggunakan QualityCheck step di SageMaker Pipelines, lihat 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"
)

EMRlangkah

Gunakan EMRlangkah Amazon SageMaker Model Building Pipelines untuk:

Untuk informasi selengkapnya tentang AmazonEMR, lihat Memulai Amazon EMR.

EMRLangkah ini mengharuskan EMRStepConfig menyertakan lokasi JAR file yang digunakan oleh EMR cluster Amazon dan argumen apa pun yang akan diteruskan. Anda juga memberikan ID EMR cluster Amazon jika Anda ingin menjalankan langkah pada EMR cluster yang sedang berjalan. Anda juga dapat meneruskan konfigurasi cluster untuk menjalankan EMR langkah pada cluster yang dibuat, dikelola, dan diakhiri untuk Anda. Bagian berikut mencakup contoh dan tautan ke buku catatan sampel yang menunjukkan kedua metode.

Luncurkan pekerjaan baru di EMR klaster Amazon yang sedang berjalan

Untuk meluncurkan pekerjaan baru di EMR klaster Amazon yang sedang berjalan, teruskan ID cluster sebagai string ke cluster_id argumenEMRStep. Contoh berikut menunjukkan prosedur ini.

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

Untuk contoh buku catatan yang memandu Anda melalui contoh lengkap, lihat SageMaker Pipelines EMR Step With Running EMR Cluster.

Luncurkan pekerjaan baru di EMR cluster Amazon baru

Untuk meluncurkan pekerjaan baru di klaster baru yang EMRStep membuat untuk Anda, berikan konfigurasi klaster Anda sebagai kamus. Kamus harus memiliki struktur yang sama dengan RunJobFlowpermintaan. Namun, jangan sertakan bidang berikut dalam konfigurasi klaster Anda:

Semua RunJobFlow argumen lain tersedia untuk digunakan dalam konfigurasi cluster Anda. Untuk detail tentang sintaks permintaan, lihat RunJobFlow.

Contoh berikut meneruskan konfigurasi cluster ke definisi EMR langkah. Ini meminta langkah untuk meluncurkan pekerjaan baru di EMR cluster baru. Konfigurasi EMR cluster dalam contoh ini mencakup spesifikasi untuk node EMR cluster primer dan inti. Untuk informasi selengkapnya tentang jenis EMR node Amazon, lihat Memahami tipe node: node primer, inti, dan tugas.

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
)

Untuk contoh buku catatan yang memandu Anda melalui contoh lengkap, lihat EMRLangkah SageMaker Pipelines Dengan Manajemen Siklus Hidup Cluster.

Langkah pekerjaan notebook

Gunakan a NotebookJobStep untuk menjalankan Job SageMaker Notebook Anda secara non-interaktif sebagai langkah pipeline. Untuk informasi selengkapnya tentang Pekerjaan SageMaker Notebook, lihatSageMaker Lowongan Notebook.

A NotebookJobStep membutuhkan minimal notebook input, image URI dan nama kernel. Untuk informasi selengkapnya tentang persyaratan langkah Job Notebook dan parameter lain yang dapat Anda atur untuk menyesuaikan langkah, lihat sagemaker.workflow.steps. NotebookJobStep.

Contoh berikut menggunakan argumen minimum untuk mendefinisikan aNotebookJobStep.

from sagemaker.workflow.notebook_job_step import NotebookJobStep


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

Langkah NotebookJobStep pipeline Anda diperlakukan sebagai pekerjaan SageMaker notebook. Akibatnya, lacak status eksekusi di dasbor pekerjaan notebook UI Studio Classic dengan menyertakan tag tertentu dengan tags argumen. Untuk detail selengkapnya tentang tag yang akan disertakan, lihatMelihat pekerjaan notebook Anda di dasbor Studio UI.

Juga, jika Anda menjadwalkan pekerjaan notebook Anda menggunakan SageMaker PythonSDK, Anda hanya dapat menentukan gambar tertentu untuk menjalankan pekerjaan notebook Anda. Untuk informasi selengkapnya, lihat Kendala gambar untuk pekerjaan notebook Python SageMaker SDK.

Langkah gagal

Gunakan a FailStep untuk menghentikan eksekusi Amazon SageMaker Model Building Pipelines bila kondisi atau status yang diinginkan tidak tercapai. Ini juga menandai bahwa eksekusi pipeline gagal. Ini FailStep juga memungkinkan Anda memasukkan pesan kesalahan khusus, yang menunjukkan penyebab kegagalan eksekusi pipeline.

Batasan untuk menggunakan FailStep

Anda dapat membuat FailStep ErrorMessage dalam bentuk string teks statis. Atau, Anda juga dapat menggunakan Parameter Pipeline operasi Gabung, atau properti langkah lainnya untuk membuat pesan kesalahan yang lebih informatif.

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

Properti langkah

Gunakan properties atribut untuk menambahkan dependensi data di antara langkah-langkah dalam pipeline. SageMaker Pipelines menggunakan dependensi data ini untuk membangun DAG dari definisi pipeline. Properti ini dapat direferensikan sebagai nilai placeholder dan diselesaikan saat runtime.

propertiesAtribut langkah SageMaker Pipelines cocok dengan objek yang dikembalikan oleh Describe panggilan untuk jenis SageMaker pekerjaan yang sesuai. Untuk setiap jenis pekerjaan, Describe panggilan mengembalikan objek respons berikut:

Untuk memeriksa properti mana yang dapat direferensikan untuk setiap tipe langkah selama pembuatan dependensi data, lihat Ketergantungan Data - Referensi Properti di Amazon Python. SageMaker SDK

Paralelisme langkah

Ketika sebuah langkah tidak bergantung pada langkah lain, itu berjalan segera setelah eksekusi pipeline. Namun, mengeksekusi terlalu banyak langkah pipeline secara paralel dapat dengan cepat menghabiskan sumber daya yang tersedia. Kontrol jumlah langkah bersamaan untuk eksekusi pipeline denganParallelismConfiguration.

Contoh berikut digunakan ParallelismConfiguration untuk mengatur batas langkah bersamaan ke lima.

pipeline.create(
    parallelism_config=ParallelismConfiguration(5),
)

Ketergantungan data antar langkah

Anda menentukan struktur Anda DAG dengan menentukan hubungan data antar langkah. Untuk membuat dependensi data antar langkah, berikan properti dari satu langkah sebagai input ke langkah lain dalam pipeline. Langkah menerima input tidak dimulai sampai setelah langkah menyediakan input selesai berjalan.

Ketergantungan data menggunakan JsonPath notasi dalam format berikut. Format ini melintasi file JSON properti. Ini berarti Anda dapat menambahkan sebanyak <property> instance yang diperlukan untuk mencapai properti bersarang yang diinginkan dalam file. Untuk informasi lebih lanjut tentang JsonPath notasi, lihat JsonPath repo.

<step_name>.properties.<property>.<property>

Berikut ini menunjukkan cara menentukan bucket Amazon S3 menggunakan ProcessingOutputConfig properti langkah pemrosesan.

step_process.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri

Untuk membuat ketergantungan data, teruskan bucket ke langkah pelatihan sebagai berikut.

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

Untuk memeriksa properti mana yang dapat direferensikan untuk setiap tipe langkah selama pembuatan dependensi data, lihat Ketergantungan Data - Referensi Properti di Amazon Python. SageMaker SDK

Ketergantungan khusus antar langkah

Saat Anda menentukan ketergantungan data, SageMaker Pipelines menyediakan koneksi data di antara langkah-langkah tersebut. Atau, satu langkah dapat mengakses data dari langkah sebelumnya tanpa langsung menggunakan SageMaker Pipelines. Dalam hal ini, Anda dapat membuat dependensi khusus yang memberi tahu SageMaker Pipelines untuk tidak memulai langkah sampai setelah langkah lain selesai berjalan. Anda membuat dependensi kustom dengan menentukan atribut step. DependsOn

Sebagai contoh, berikut ini mendefinisikan langkah C yang dimulai hanya setelah kedua langkah A dan langkah B selesai berjalan.

{
  'Steps': [
    {'Name':'A', ...},
    {'Name':'B', ...},
    {'Name':'C', 'DependsOn': ['A', 'B']}
  ]
}

SageMaker Pipelines melempar pengecualian validasi jika ketergantungan akan membuat ketergantungan siklik.

Contoh berikut membuat langkah pelatihan yang dimulai setelah langkah pemrosesan selesai berjalan.

processing_step = ProcessingStep(...)
training_step = TrainingStep(...)

training_step.add_depends_on([processing_step])

Contoh berikut membuat langkah pelatihan yang tidak dimulai sampai dua langkah pemrosesan yang berbeda selesai berjalan.

processing_step_1 = ProcessingStep(...)
processing_step_2 = ProcessingStep(...)

training_step = TrainingStep(...)

training_step.add_depends_on([processing_step_1, processing_step_2])

Berikut ini menyediakan cara alternatif untuk membuat ketergantungan kustom.

training_step.add_depends_on([processing_step_1])
training_step.add_depends_on([processing_step_2])

Contoh berikut membuat langkah pelatihan yang menerima masukan dari satu langkah pemrosesan dan menunggu langkah pemrosesan yang berbeda untuk selesai berjalan.

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

Contoh berikut menunjukkan bagaimana untuk mengambil daftar string dari dependensi kustom langkah.

custom_dependencies = training_step.depends_on

Gunakan gambar khusus dalam satu langkah

Anda dapat menggunakan salah satu gambar SageMaker Deep Learning Container yang tersedia saat membuat langkah dalam pipeline.

Anda juga dapat menggunakan wadah Anda sendiri dengan langkah-langkah pipa. Karena Anda tidak dapat membuat gambar dari dalam Studio Classic, Anda harus membuat gambar menggunakan metode lain sebelum menggunakannya dengan SageMaker Pipelines.

Untuk menggunakan penampung Anda sendiri saat membuat langkah-langkah untuk pipeline Anda, sertakan gambar URI dalam definisi estimator. Untuk informasi selengkapnya tentang menggunakan kontainer Anda sendiri SageMaker, lihat Menggunakan Kontainer Docker dengan SageMaker.