Pipeline-Schritte - Amazon SageMaker
SchritttypenStep-EigenschaftenSchrittparallelitätDatenabhängigkeit zwischen SchrittenBenutzerdefinierte Abhängigkeiten zwischen SchrittenVerwenden Sie ein benutzerdefiniertes Bild in einem Schritt

Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Pipeline-Schritte

SageMaker Pipelines bestehen aus Schritten. Diese Schritte definieren die Aktionen, die die Pipeline durchführt, und die Beziehungen zwischen Schritten, die Eigenschaften verwenden.

Schritttypen

Im Folgenden werden die Anforderungen jedes Schritttyps beschrieben und eine Beispielimplementierung des Schrittes dargestellt. Dies sind keine funktionalen Implementierungen, da sie nicht die benötigten Ressourcen und Eingaben bereitstellen. Ein Tutorial, das diese Schritte implementiert, finden Sie unterErstellen und Verwalten SageMaker Pipelinesaus.

Amazon SageMaker Modellbau-Pipelines unterstützen die folgenden Schritttypen:

-Schrittverarbeitung

Mit einem Verarbeitungsschritt erstellen Sie einen Verarbeitungsauftrag für die Datenverarbeitung. Weitere Informationen zur Verarbeitung von Aufträgen finden Sie unterVerarbeiten von Daten und Auswerten von Modellenaus.

Ein Verarbeitungsschritt erfordert einen Prozessor, ein Python-Skript, das den Verarbeitungscode definiert, Ausgaben für die Verarbeitung und Auftragsargumente. Im folgenden Beispiel wird gezeigt, wie einProcessingStepDefinition. Weitere Informationen über Anforderungen für die Verarbeitung von Schritten finden Sie unter derSageMaker.workflow.steps.ProcessingStep-Dokumentation.

from sagemaker.sklearn.processing import SKLearnProcessor

sklearn_processor = SKLearnProcessor(framework_version='0.20.0',
                                     role=<role>,
                                     instance_type='ml.m5.xlarge',
                                     instance_count=1)
from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep

step_process = ProcessingStep(
    name="AbaloneProcess",
    processor=sklearn_processor,
    inputs=[
      ProcessingInput(source=<input_data>, destination="/opt/ml/processing/input"),
    ],
    outputs=[
        ProcessingOutput(output_name="train", source="/opt/ml/processing/train"),
        ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"),
        ProcessingOutput(output_name="test", source="/opt/ml/processing/test")
    ],
    code="abalone/preprocessing.py"
)

Übergeben von Laufzeitparametern

Sie können Laufzeitparameter an einen Verarbeitungsschritt übergeben, indem Sie dieget_run_args-Methode desAmazon SageMaker Python SDKaus. Auf diese Weise können Sie außerdem Prozessoren verwendenSKLearnProcessor, wie beispielsweisePySparkProcessorundSparkJarProcessoraus.

Das folgende Beispiel zeigt, wie Laufzeitparameter von einem übergeben werden PySpark Prozessor zu einemProcessingStepaus.

from sagemaker.spark.processing import PySparkProcessor

pyspark_processor = PySparkProcessor(framework_version='2.4',
                                     role=<role>,
                                     instance_type='ml.m5.xlarge',
                                     instance_count=1)
from sagemaker.processing import ProcessingInput, ProcessingOutput

run_args = pyspark_processor.get_run_args(
    "preprocess.py",
    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")
    ],
    arguments=None
)

from sagemaker.workflow.steps import ProcessingStep

step_process = ProcessingStep(
    name="AbaloneProcess",
    processor=pyspark_processor,
    inputs=run_args.inputs,
    outputs=run_args.outputs,
    job_arguments=run_args.arguments,
    code=run_args.code
)

Schulungsschritt

Mit einem Trainingsschritt erstellen Sie einen Trainingsauftrag zum Trainieren eines Modells. Weitere Informationen zu Schulungsaufträgen finden Sie unterTrainieren eines Modells mit Amazon SageMakeraus.

Ein Schulungsschritt erfordert einen Schätzer sowie Schulungs- und Validierungsdateneingaben. Im folgenden Beispiel wird gezeigt, wie einTrainingStepDefinition. Weitere Informationen zu den Anforderungen für Schulungsschritte finden Sie im AbschnittSageMaker.workflow.steps.TrainingStep-Dokumentation.

from sagemaker.inputs import TrainingInput
from sagemaker.workflow.steps import TrainingStep

step_train = TrainingStep(
    name="TrainAbaloneModel",
    estimator=xgb_train,
    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"
        )
    }
)

Optimierungsschritt

Mit einem Optimierungsschritt erstellen Sie einen Hyperparameteroptimierungsauftrag, der auch als Hyperparameteroptimierung (HPO) bezeichnet wird. Ein Hyperparameter-Tuning-Job führt mehrere Schulungsjobs aus, von denen jeder eine Modellversion erzeugt. Weitere Informationen zu Hyperparameter-Tuning finden Sie unterDurchführen der automatischen Modelloptimierung mit SageMakeraus.

Der Tuning-Job ist mit dem SageMaker experimentieren Sie für die Pipeline, wobei die Ausbildungsplätze als Prüfungen geschaffen wurden. Weitere Informationen finden Sie unter Experimente Integration.

Ein Tuning-Schritt erfordert einenHyperParameterTunerund Schulungseingaben. Sie können frühere Tuning-Jobs umschulen, indem Sie diewarm_start_config-Parameter desHyperparameterTuneraus. Weitere Informationen zur Hyperparameter-Tuning und zum Warmstart finden Sie unterDurchführen eines Hyperparameter-Optimierungsauftrags mit Warmstartaus.

Sie verwenden dasget_top_model_s3_uri-Methode desSageMaker.workflow.steps.tuningStep-Klasse, um das Modellartefakt aus einer der leistungsstärksten Modellversionen zu erhalten. Für ein Notebook, das zeigt, wie Sie einen Tuning-Schritt in einer SageMaker Pipeline, siehesagemaker-pipelines-tuning-step.ipynbaus.

Wichtig

Optimierungsschritte wurden in Amazon eingeführt SageMaker Python SDK v2.48.0 und Amazon SageMaker Studio v3.8.0. Sie müssen Studio aktualisieren, bevor Sie einen Tuning-Schritt verwenden, sonst wird die Pipeline-DAG nicht angezeigt. Weitere Informationen zu So aktualisieren Sie Studio unter.Aktualisieren und Herunterfahren SageMaker Studioaus.

Im folgenden Beispiel wird gezeigt, wie einTuningStepDefinition.

from sagemaker.tuner import HyperparameterTuner
from sagemaker.inputs import TrainingInput
from sagemaker.workflow.steps import TuningStep
    
step_tuning = TuningStep(
    name = "HPTuning",
    tuner = HyperparameterTuner(...),
    inputs = TrainingInput(s3_data=input_path)
)

Holen Sie sich die beste Modellversion

Im folgenden Beispiel wird gezeigt, wie Sie die beste Modellversion aus dem Tuning-Job mit der Option erhalten.get_top_model_s3_uri-Methode. Höchstens sind die 50 leistungsstärksten Versionen erhältlich nachHyperparameterTuningJobObjectiveaus. Dietop_kArgument ist ein Index in die Versionen, wotop_k=0ist die Version mit der besten Leistung undtop_k=49ist die Version mit der schlechtesten Leistung.

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

Weitere Informationen zu Anforderungen für Optimierungsschritte finden Sie unter derSageMaker.workflow.steps.tuningStep-Dokumentation.

CreateModel Step

Du benutzt einCreateModelSchritt zum Erstellen eines SageMaker -Modell. Weitere Informationen zu SageMaker Modelle, sieheTrainieren eines Modells mit Amazon SageMakeraus.

Ein Schritt zum Erstellen eines Modells erfordert Modellartefakte und Informationen über die SageMaker Instanztyp, den Sie zum Erstellen des Modells verwenden müssen. Im folgenden Beispiel wird gezeigt, wie einCreateModelSchrittdefinition. Weitere Informationen zuCreateModelSchrittanforderungen finden Sie unter.SageMaker.workflow.steps.CreateModelStep-Dokumentation.

from sagemaker.workflow.steps import CreateModelStep

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

RegisterModel Step

Du benutzt einRegisterModelschritt zur Registrierung einesSageMaker.model.modeloder einSageMaker.pipeline.PipelineModelmit dem Amazon SageMaker -Modellregistrierungsstelle. EINPipelineModelstellt eine Inferenz-Pipeline dar, bei der es sich um ein Modell handelt, das aus einer linearen Folge von Containern besteht, die Inferenzanforderungen verarbeiten.

Weitere Informationen zum Registrieren eines Modells finden Sie unterRegistrieren und Bereitstellen von Modellen mit Modellregistrierungaus. Weitere Informationen zuRegisterModelSchrittanforderungen finden Sie unter.SageMaker.workflow.step_Collections.RegisterModel-Dokumentation.

Im folgenden Beispiel wird gezeigt, wie einRegisterModelSchritt, der ein registriertPipelineModelaus.

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

WennmodelDer Register-Modellschritt benötigt einen Schätzwert, wie in dem folgenden Beispiel gezeigt.

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 Step

Sie verwenden einen Transformationsschritt für die Batch-Transformation, um Inferenzen auf ein ganzes Dataset auszuführen. Weitere Informationen zur Stapeltransformation finden Sie unterFühren Sie Folgendes aus:Stapeltransformationen mit Inferenz-Pipelinesaus.

Ein Transformationsschritt erfordert einen Transformator und die Daten, auf denen die Batch-Transformation ausgeführt werden soll. Im folgenden Beispiel wird gezeigt, wie einTransformSchrittdefinition. Weitere Informationen zuTransformSchrittanforderungen finden Sie unter.SageMaker.workflow.steps.TransformStep-Dokumentation.

from sagemaker.inputs import TransformInput
from sagemaker.workflow.steps import TransformStep

step_transform = TransformStep(
    name="AbaloneTransform",
    transformer=transformer,
    inputs=TransformInput(data=batch_data)
)

Schritt „Zustand“

Verwenden Sie einen Bedingungsschritt, um den Zustand der Schritteigenschaften zu bewerten, um zu beurteilen, welche Maßnahmen als nächstes in der Pipeline ergriffen werden sollen.

Ein Bedingungsschritt erfordert eine Liste von Bedingungen, eine Liste von Schritten, die ausgeführt werden sollen, wenn die Bedingung ausgewertet wirdtrueund eine Liste von Schritten, die ausgeführt werden sollen, wenn die Bedingung ausgewertet wirdfalseaus. Im folgenden Beispiel wird gezeigt, wie einConditionSchrittdefinition. Weitere Informationen zuConditionSchrittanforderungen finden Sie unter.SageMaker.workflow.Condition_Step.ConditionStep-Dokumentation.

Einschränkungen

  • SageMaker Pipelines unterstützt die Verwendung von verschachtelten Zustandsschritten nicht. Sie können einen Bedingungsschritt nicht als Eingabe für einen anderen Bedingungsschritt übergeben.

  • Ein Bedingungsschritt kann in beiden Zweigen keine identischen Schritte verwenden. Wenn Sie in beiden Zweigen dieselbe Schrittfunktionalität benötigen, duplizieren Sie den Schritt und geben Sie ihm einen anderen Namen.

from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo
from sagemaker.workflow.condition_step import (
    ConditionStep,
    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=[]
)

Rückrufschritt

Du benutzt einCallbackSchritt zur Einbeziehung zusätzlicher Prozesse undAWSDienste in Ihren Workflow, die nicht direkt von Amazon bereitgestellt werden SageMaker Modellentwicklungs-Pipelines. Wenn einCallbackstep läuft, das folgende Verfahren tritt auf:

  • SageMaker Pipelines sendet eine Nachricht an eine vom Kunden angegebene Amazon Simple Queue Service (Amazon SQS) -Warteschlange. Die Nachricht enthält eine SageMaker Pipelines — Generiertes Token und eine vom Kunden bereitgestellte Liste von Eingabeparametern. Nach dem Senden der Nachricht SageMaker Pipelines warten auf eine Antwort des Kunden.

  • Der Kunde ruft die Nachricht aus der Amazon-SQS-Warteschlange ab und startet seinen benutzerdefinierten Prozess.

  • Wenn der Prozess abgeschlossen ist, ruft der Kunde eine der folgenden APIs an und sendet die SageMaker Pipelines — Generiertes Token:

  • Der API-Aufruf verursacht SageMaker Pipelines, um entweder den Pipeline-Prozess fortzusetzen oder den Prozess fehlschlagen zu können.

Weitere Informationen zuCallbackSchrittanforderungen finden Sie unter.SageMaker.workflow.callback_step.callbackStep-Dokumentation. Eine Komplettlösung finden Sie unterErweitern SageMaker Pipelines zur Einbeziehung benutzerdefinierter Schritte mit Rückrufschrittenaus.

Wichtig

CallbackSchritte wurden in Amazon eingeführt SageMaker Python SDK v2.45.0 und Amazon SageMaker Studio v3.6.2. Sie müssen Studio aktualisieren, bevor Sie eineCallbackstep oder die Pipeline DAG wird nicht angezeigt. Weitere Informationen zu So aktualisieren Sie Studio unter.Aktualisieren und Herunterfahren SageMaker Studioaus.

Die folgende Stichprobe zeigt eine Umsetzung des vorangegangenen Verfahrens.

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

Verhalten stoppen

Ein Pipeline-Prozess hört nicht auf, während einCallbackSpringen zu.

Wenn du anrufstStopPipelineExecutionbei einem Pipeline-Prozess mit einem laufendenCallbackSchritt, SageMaker Pipelines senden eine zusätzliche Amazon SQS SQS-Nachricht an die angegebene SQS-Warteschlange. Der Text der SQS-Nachricht enthält eineStatusFeld, aufStoppingaus. Das folgende zeigt ein Beispiel für einen SQS-Nachrichtentext.

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

Sie sollten Ihrem Amazon SQS SQS-Nachrichtenverbraucher Logik hinzufügen, um nach Erhalt der Nachricht alle erforderlichen Maßnahmen (z. B.SendPipelineExecutionStepSuccessoderSendPipelineExecutionStepFailureaus.

Nur wenn SageMaker Pipelines erhalten einen dieser Aufrufe, stoppt der Pipeline-Prozess.

Lambda-Schritt

Sie verwenden einen Lambda-Schritt, um eineAWS LambdaFunktion. Sie können eine vorhandene Lambda-Funktion ausführen oder SageMaker kann eine neue Lambda-Funktion erstellen und ausführen. Für ein Notizbuch, das zeigt, wie man einen Lambda-Schritt in einem SageMaker Pipeline, sieheSagemacher-Pipelines-lambda-step.ipynbaus.

Wichtig

Lambda-Schritte wurden in Amazon eingeführt SageMaker Python SDK v2.51.0 und Amazon SageMaker Studio v3.9.1. Sie müssen Studio aktualisieren, bevor Sie einen Lambda-Schritt verwenden, sonst wird die Pipeline-DAG nicht angezeigt. Weitere Informationen zu So aktualisieren Sie Studio unter.Aktualisieren und Herunterfahren SageMaker Studioaus.

SageMaker bietetSagemaker.lambda_Helper.LambdaKlasse zum Erstellen, Aktualisieren, Aufrufen und Löschen von Lambda-Funktionen.Lambdahat die folgende Signatur.

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

DieSageMaker.workflow.lambda_step.lambdaStepKlasse hat einelambda_funcArgumenttypLambdaaus. Um eine vorhandene Lambda-Funktion aufzurufen, müssen Sie nur den Amazon-Ressourcennamen (ARN) der Funktion anfunction_arnaus. Wenn Sie keinen Wert für angebenfunction_arnangeben, müssen Sie angebenhandlerund einer der folgenden Aufgaben:

  • zipped_code_dir— Der Pfad der gezippten Lambda-Funktion

    s3_bucket— Amazon S3 S3-Bucket wozipped_code_dirsoll hochgeladen werden

  • script— Der Pfad der Lambda-Funktionsskriptdatei

Im folgenden Beispiel wird gezeigt, wie einLambdaSchrittdefinition, die eine vorhandene Lambda-Funktion aufruft.

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

Im folgenden Beispiel wird gezeigt, wie einLambdaSchrittdefinition, die eine Lambda-Funktion mit einem Lambda-Funktionsskript erstellt und aufruft.

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

Eingaben und Ausgaben

Wenn IhreLambdaFunktion hat Ein- oder Ausgänge, diese müssen auch in IhremLambdaSchritt.

Bei der Definition desLambdaSchritt,inputsEs muss ein Wörterbuch mit Schlüssel-Wert-Paaren sein. Jeder Wert desinputswörterbuch muss ein primitiver Typ sein (String, Integer oder Float). Verschachtelte Objekte werden nicht unterstützt. Wenn undefiniert bleibt, wird derinputsWert ist standardmäßigNoneaus.

DieoutputsDer Wert muss eine Liste von Schlüsseln sein. Diese Schlüssel beziehen sich auf ein Wörterbuch, das in der Ausgabe desLambdaFunktion. LIKEinputsmüssen diese Schlüssel primitive Typen sein, und verschachtelte Objekte werden nicht unterstützt.

Timeout und Stoppverhalten

DieLambdaKlasse hat einetimeout-Argument, das die maximale Zeit angibt, die die Lambda-Funktion ausführen kann. Der Standardwert beträgt 120 Sekunden bei einem Maximalwert von 10 Minuten. Wenn die Lambda-Funktion ausgeführt wird, wenn das Timeout erreicht ist, schlägt der Lambda-Schritt fehl, die Lambda-Funktion wird jedoch weiterhin ausgeführt.

Ein Pipeline-Prozess kann nicht gestoppt werden, während ein Lambda-Schritt ausgeführt wird, da die vom Lambda-Schritt aufgerufene Lambda-Funktion nicht gestoppt werden kann. Wenn Sie versuchen, den Prozess zu stoppen, während die Lambda-Funktion ausgeführt wird, wartet die Pipeline darauf, dass die Lambda-Funktion beendet ist oder bis das Timeout getroffen wird, je nachdem, was zuerst auftritt, und stoppt dann. Wenn die Lambda-Funktion beendet ist, lautet der Pipeline-ProzessstatusStoppedaus. Wenn das Timeout erreicht ist, lautet der Pipeline-ProzessstatusFailedaus.

ClarifyCheck Schritt

Sie können dasClarifyCheckSchritt zur Durchführung von Baseline-Drift-Checks gegenüber früheren Baselines für Bias-Analyse und Modellerklärbarkeit. Auf diese Weise können Sie und generierenRegistrieren Sie Ihre Baselinesfür Drift-Check imRegisterModelzusammen mit dem Modell wurden die Pipelines in der Modellbauphase durchtrainiert. Diese Baselines für Drift-Check können von Amazon verwendet werden SageMaker Model Monitor für Ihre Modellendpunkte, damit Sie keineGrundlinieVorschlag getrennt. DieClarifyCheckstep kann auch Baselines für Drift-Check aus der Modellregistrierung ziehen. DieClarifyCheckstep nutzt den Amazonas SageMaker Klären Sie einen vorgefertigten Container, der eine Reihe von Modellüberwachungsfunktionen bereitstellt, einschließlich Einschränkungsvorschlag und Einschränkungsvalidierung anhand einer bestimmten Baseline. Weitere Informationen finden Sie unterErste Schritte mit einem SageMaker Behältnis klärenaus.

Konfigurieren von ClarifyCheck Schritt

Sie können denClarifyCheckSchritt, um jedes Mal, wenn er in einer Pipeline verwendet wird, nur einen der folgenden Prüftypen durchzuführen.

  • Überprüfung der Datenbias-Kontrolle

  • Modellbias-Check

  • Prüfung der Erklärbarkeit des Modells

Sie tun dies, indem Sie die Option festlegenclarify_check_config-Parameter mit einem der folgenden Prüftypwerte:

  • DataBiasCheckConfig

  • ModelBiasCheckConfig

  • ModelExplainabilityCheckConfig

DieClarifyCheckstep startet einen Verarbeitungsauftrag, der SageMaker Klären Sie vorgefertigte Container und erfordert dedizierteKonfigurationen für den Check- und Verarbeitungsauftragaus.ClarifyCheckConfigundCheckJobConfigsind Hilfsfunktionen für diese Konfigurationen, die darauf ausgerichtet sind, wie die SageMaker Klären Sie den Verarbeitungsauftrag für die Überprüfung der Modellverzerrung, der Datenverzerrung oder der Erklärbarkeit von Modellen. Weitere Informationen finden Sie unter Führen Sie Folgendes aus: SageMaker Verarbeitungsjobs für Bias-Analyse und Erklärbarkeit klären.

Steuern des Schrittverhaltens für Drift-Check

DieClarifyCheckstep benötigt die folgenden zwei booleschen Flags, um sein Verhalten zu steuern:

  • skip_check: Dieser Parameter gibt an, ob die Drift-Prüfung gegenüber der vorherigen Baseline übersprungen wird oder nicht. Wenn es auf eingestellt istFalsemuss die vorherige Baseline des konfigurierten Prüftyps verfügbar sein.

  • register_new_baseline: Dieser Parameter gibt an, ob über die Step-Eigenschaft auf eine neu berechnete Baseline zugegriffen werden kannBaselineUsedForDriftCheckConstraintsaus. Wenn es auf eingestellt istFalsemuss auch die vorherige Baseline des konfigurierten Prüftyps verfügbar sein. Auf diese Option kann über dieBaselineUsedForDriftCheckConstraintsEigentum.

Weitere Informationen finden Sie unter Baseline-Berechnung, Drift-Erkennung und Lebenszyklus mit ClarifyCheck und QualityCheck Schritte in Amazon SageMaker Pipelines für die Modellerstellung.

Arbeiten mit Baselines

Optional können Sie diemodel_package_group_nameum die vorhandene Baseline und dieClarifyCheckSchritt zieht denDriftCheckBaselinesauf das neueste genehmigte Modellpaket in der Modellpaketgruppe. Oder Sie können eine vorherige Baseline über diesupplied_baseline_constraints-Parameter. Wenn Sie beide angebenmodel_package_group_nameund diesupplied_baseline_constraints, derClarifyCheckstep verwendet die Baseline, die durch diesupplied_baseline_constraints-Parameter.

Weitere Informationen zur Verwendung der RegionClarifyCheckSchrittanforderungen finden Sie unter.SageMaker.workflow.steps.ClarifyCheckStepimAmazon SageMaker SageMaker SDK für Pythonaus. Für einen Amazonas SageMaker Studio-Notizbuch, das zeigt, wie man benutztClarifyCheckSpringen zu SageMaker Pipelines, siehesagemacher-Pipeline-Modell-Monitor-Clarify-steps.ipynbaus.

Beispiel Erstellen einesClarifyCheckSchritt zur Überprüfung der Datenverzerrung

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 Schritt

Sie können dasQualityCheckSchritt zu führenBasisvorschlägeund Drift prüft mit einer früheren Baseline auf Datenqualität oder Modellqualität in einer Pipeline. Auf diese Weise können Sie und generierenRegistrieren Sie Ihre Baselinesfür Drift-Check imRegisterModel StepSchritt zusammen mit dem Modell, das in der Modellbauphase durch die Pipelines trainiert wurde. Model Monitor kann diese Baselines für die Drift-Überprüfung Ihrer Modellendpunkte verwenden, sodass Sie keinen Basisvorschlag separat ausführen müssen. DieQualityCheckstep kann auch Baselines für Drift-Check aus der Modellregistrierung ziehen. DieQualityCheckstep nutzt den Amazonas SageMaker Vorgefertigter Container, der über eine Reihe von Modellüberwachungsfunktionen verfügt, darunter Einschränkungsvorschlag, Statistikgenerierung und Einschränkungsvalidierung anhand einer Baseline. Weitere Informationen finden Sie unter Amazon SageMaker Vorgefertigter Container Model Monitor.

Konfigurieren von QualityCheck Schritt

Sie können denQualityCheckSchritt, um jedes Mal, wenn er in einer Pipeline verwendet wird, nur einen der folgenden Prüftypen durchzuführen.

  • Überprüfung der Datenqualität

  • Modellqualitätsprüfung

Sie tun dies, indem Sie die Option festlegenquality_check_config-Parameter mit einem der folgenden Prüftypwerte:

  • DataQualityCheckConfig

  • ModelQualityCheckConfig

DieQualityCheckstep startet einen Verarbeitungsauftrag, der den vorgefertigten Model Monitor-Container ausführt und dedizierte Konfigurationen für den Prüf- und Verarbeitungsauftrag benötigt. DieQualityCheckConfigundCheckJobConfigsind Hilfsfunktionen für diese Konfigurationen, die darauf ausgerichtet sind, wie Model Monitor eine Basislinie für die Überwachung der Modellqualität oder Datenqualität erstellt. Weitere Informationen zu den Baseline-Vorschlägen für den Modellmonitor finden Sie unterErstellen einer BaselineundErstellen einer Modellqualitäts-Baselineaus.

Steuern des Schrittverhaltens für Drift-Check

DieQualityCheckstep benötigt die folgenden zwei booleschen Flags, um sein Verhalten zu steuern:

  • skip_check: Dieser Parameter gibt an, ob die Drift-Prüfung gegenüber der vorherigen Baseline übersprungen wird oder nicht. Wenn es auf eingestellt istFalsemuss die vorherige Baseline des konfigurierten Prüftyps verfügbar sein.

  • register_new_baseline: Dieser Parameter gibt an, ob über Schritteigenschaften auf eine neu berechnete Baseline zugegriffen werden kannBaselineUsedForDriftCheckConstraintsundBaselineUsedForDriftCheckStatisticsaus. Wenn es auf eingestellt istFalsemuss auch die vorherige Baseline des konfigurierten Prüftyps verfügbar sein. Auf diese können Sie über die Option zugreifenBaselineUsedForDriftCheckConstraintsundBaselineUsedForDriftCheckStatisticsimmobilien.

Weitere Informationen finden Sie unter Baseline-Berechnung, Drift-Erkennung und Lebenszyklus mit ClarifyCheck und QualityCheck Schritte in Amazon SageMaker Pipelines für die Modellerstellung.

Arbeiten mit Baselines

Sie können eine vorherige Baseline direkt über diesupplied_baseline_statisticsundsupplied_baseline_constraints-Parameter oder Sie können einfach denmodel_package_group_nameund dieQualityCheckSchritt zieht denDriftCheckBaselinesauf das neueste genehmigte Modellpaket in der Modellpaketgruppe. Wenn Sie diemodel_package_group_name, dersupplied_baseline_constraints, undsupplied_baseline_statistics, derQualityCheckstep verwendet die Baseline, die vonsupplied_baseline_constraintsundsupplied_baseline_statisticsüber den Schecktyp desQualityCheckSchritt du rennst.

Weitere Informationen zur Verwendung der RegionQualityCheckSchrittanforderungen finden Sie unter.SageMaker.workflow.steps.QualityCheckStepimAmazon SageMaker SageMaker SDK für Pythonaus. Für einen Amazonas SageMaker Studio-Notizbuch, das zeigt, wie man benutztQualityCheckSpringen zu SageMaker Pipelines, siehesagemacher-Pipeline-Modell-Monitor-Clarify-steps.ipynbaus.

Beispiel Erstellen einesQualityCheckSchritt für Datenqualitätsprüfung

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 EMR Step

Sie können Amazon verwenden SageMaker Pipelines für die ModellerstellungAmazon EMRSchritt zu verarbeitenEMR-Schrittean einen laufenden Amazon EMR-Cluster. Weitere Informationen finden Sie unterErste Schritte mit Amazon EMRaus.

Der Amazon EMR-Schritt erfordert eineEMRStepConfigmit dem Amazon S3 S3-Speicherort des JAR, der vom Amazon EMR-Cluster verwendet werden soll, und alle zu übergebenden Argumente sowie die Amazon EMR-Cluster-ID.

Anmerkung

  • Amazon EMR-Schritte erfordern, dass die an Ihre Pipeline übergebene Rolle zusätzliche Berechtigungen besitzt. Sie sollten dieAWSVerwaltete Richtlinie: AmazonSageMakerPipelinesIntegrationsauf Ihre Pipeline-Rolle oder stellen Sie sicher, dass die Rolle die Berechtigungen in dieser Richtlinie enthält.

  • Amazon EMR in EKS wird nicht unterstützt.

  • Sie können einen EMR-Schritt nur für ein Cluster ausführen, das sich in einem der folgenden Zustände befindet:STARTING,BOOTSTRAPPING,RUNNING, oderWAITINGaus.

  • Sie können höchstens 256 EMR-Schritte inPENDINGauf einem EMR-Cluster; EMR-Schritte, die über dieses Limit hinaus gesendet werden, führen dazu, dass die Pipeline-Ausführung fehlschlägt. Sie können in Erwägung ziehenRichtlinie für Pipeline-Schritte wiederholenaus.

Beispiel Erstellen Sie eine Amazon EMR-Schrittdefinition, die einen neuen Job in einem EMR-Cluster startet.

from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_config = EMRStepConfig(
    jar="s3://path/to/jar/MyJar.jar", # required, S3 path to jar
    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-1YONHTCP3YZKC", # required
    step_config=emr_config, # required
    display_name="My EMR Step",
    description="Pipeline step to execute EMR job"
)

Schritt scheitern

Du benutzt einFailStepum einen Amazonas zu stoppen SageMaker Ausführung von Modellbau-Pipelines, wenn eine gewünschte Bedingung oder Status nicht erreicht wird, und um die Ausführung dieser Pipeline als fehlgeschlagen zu markieren. DieFailStepermöglicht es Ihnen auch, eine benutzerdefinierte Fehlermeldung einzugeben, die die Ursache für den Ausführungsfehler der Pipeline angibt.

Anmerkung

Wenn einFailStepund andere Pipeline-Schritte werden gleichzeitig ausgeführt, die Pipeline wird erst beendet, wenn alle gleichzeitigen Schritte abgeschlossen sind.

Einschränkungen bei der Verwendung vonFailStep

  • Sie können keinFailStepzumDependsOnListe anderer Schritte. Weitere Informationen finden Sie unter Benutzerdefinierte Abhängigkeiten zwischen Schritten.

  • Andere Schritte können nicht aufFailStepaus. Es istimmerder letzte Schritt in der Ausführung einer Pipeline.

  • Sie können eine Pipeline-Ausführung nicht erneut versuchen, die mit einer endetFailStepaus.

Sie können das erstellenFailStep ErrorMessagein Form einer statischen Textzeichenfolge. Alternativ können Sie auch verwendenPipeline-ParametereinBeitretenBetrieb oder anderestep-Eigenschaftenum eine informativere Fehlermeldung zu erstellen.

Das folgende Beispielcode-Snippet verwendet einFailStepmit einemErrorMessagekonfiguriert mit Pipeline-Parametern und einemJoinverwenden.

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

Step-Eigenschaften

Dieproperties-Attribut wird verwendet, um Datenabhängigkeiten zwischen Schritten in der Pipeline hinzuzufügen. Diese Datenabhängigkeiten werden dann von SageMaker Pipelines zum Konstruieren der DAG aus der Pipeline-Definition. Diese Eigenschaften können als Platzhalterwerte referenziert werden und werden zur Laufzeit aufgelöst.

DiepropertiesAttribut einer SageMaker Pipelines Schritt stimmt mit dem Objekt überein, das von aDescriberufen Sie nach dem entsprechenden SageMaker Auftragstyp Für jeden Jobtyp ist derDescribecall gibt das folgende Antwortobjekt zurück:

Schrittparallelität

Wenn ein Schritt nicht von einem anderen Schritt abhängt, wird er sofort nach der Pipeline-Ausführung ausgeführt. Die parallel Ausführung zu vieler Pipeline-Schritte kann jedoch schnell verfügbare Ressourcen ausschöpfen. Steuern Sie die Anzahl der gleichzeitigen Schritte für eine Pipeline-Ausführung mitParallelismConfigurationaus.

Im folgenden Beispiel wird verwendetParallelismConfigurationum das Limit für gleichzeitige Schritte auf fünf festzulegen.

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

Datenabhängigkeit zwischen Schritten

Sie definieren die Struktur Ihrer DAG, indem Sie die Datenbeziehungen zwischen Schritten angeben. Um Datenabhängigkeiten zwischen Schritten zu erstellen, übergeben Sie die Eigenschaften eines Schritts als Eingabe an einen anderen Schritt in der Pipeline. Der Schritt, der die Eingabe erhält, wird erst gestartet, nachdem der Schritt, der die Eingabe bereitstellt, ausgeführt wurde.

Eine Datenabhängigkeit verwendet JsonPath -Notation im folgenden Format. Dieses Format durchläuft die JSON-Eigenschaftsdatei, was bedeutet, dass Sie so viele anhängen können<property>Instanzen nach Bedarf, um die gewünschte verschachtelte Eigenschaft in der Datei zu erreichen. Weitere Informationen zu JsonPath Notation, siehe dieJSONPath-Repoaus.

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

Im Folgenden wird gezeigt, wie ein Amazon-S3-Bucket unter Verwendung desProcessingOutputConfigEigentum eines Verarbeitungsschritts.

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

Um die Datenabhängigkeit zu erstellen, übergeben Sie den Bucket wie folgt an einen Trainingsschritt.

step_train = TrainingStep(
    name="CensusTrain",
    estimator=sklearn_train,
    inputs=TrainingInput(
        s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
            "train_data"].S3Output.S3Uri
    )
)

Benutzerdefinierte Abhängigkeiten zwischen Schritten

Wenn Sie eine Datenabhängigkeit angeben, SageMaker Pipelines stellen die Datenverbindung zwischen den Schritten bereit. Alternativ kann ein Schritt aus einem vorherigen Schritt auf die Daten zugreifen, ohne direkt zu verwenden SageMaker Pipelines. In diesem Fall können Sie eine benutzerdefinierte Abhängigkeit erstellen, die sagt SageMaker Pipelines, um einen Schritt erst zu starten, nachdem ein weiterer Schritt beendet ist. Sie erstellen eine benutzerdefinierte Abhängigkeit, indem Sie dieDependsOnAttribut.

Als Beispiel definiert das Folgende einen SchrittCdas beginnt erst nach beiden SchrittenASpringen zuBfertig laufen.

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

SageMaker Pipelines löst eine Validierungsausnahme aus, wenn die Abhängigkeit eine zyklische Abhängigkeit erzeugen würde.

Im folgenden Beispiel wird ein Trainingsschritt erstellt, der beginnt, nachdem ein Verarbeitungsschritt ausgeführt wird.

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

training_step.add_depends_on([processing_step])

Im folgenden Beispiel wird ein Trainingsschritt erstellt, der erst beginnt, wenn zwei verschiedene Verarbeitungsschritte ausgeführt werden.

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

training_step = TrainingStep(...)

training_step.add_depends_on([processing_step_1, processing_step_2])

Im Folgenden finden Sie eine alternative Möglichkeit, die benutzerdefinierte Abhängigkeit zu erstellen.

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

Im folgenden Beispiel wird ein Trainingsschritt erstellt, der Eingaben aus einem Verarbeitungsschritt erhält und darauf wartet, dass ein anderer Verarbeitungsschritt beendet wird.

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

Das folgende Beispiel zeigt, wie Sie eine Zeichenfolgenliste der benutzerdefinierten Abhängigkeiten eines Schritts abrufen.

custom_dependencies = training_step.depends_on

Verwenden Sie ein benutzerdefiniertes Bild in einem Schritt

Sie können jeden der verfügbaren SageMaker verwendenDeep Learning ContainerWenn Sie einen Schritt in Ihrer Pipeline erstellen.

Sie können auch einen Schritt mit erstellen SageMaker Amazon S3 S3-Anwendungen. EIN SageMaker Amazon S3 S3-Anwendung ist ein tar.gz Bundle mit einem oder mehreren Python-Skripten, die in diesem Bundle ausgeführt werden können. Weitere Informationen zur Bündelung von Anwendungspaketen finden Sie unter.Bereitstellen einer direkten Bereitstellung von Modell-Artefaktenaus.

Sie können Ihren eigenen Container auch mit Pipeline-Schritten verwenden. Weil Sie kein Bild aus Amazon erstellen können SageMaker Studio, Sie müssen Ihr Bild mit einer anderen Methode erstellen, bevor Sie es mit Amazon verwenden SageMaker Modellentwicklungs-Pipelines.

Um beim Erstellen der Schritte für Ihre Pipeline Ihren eigenen Container zu verwenden, nehmen Sie den Bild-URI in die Schätzdefinition ein. Weitere Informationen zur Verwendung eines eigenen Containers mit SageMaker finden Sie unterVerwenden von Docker-Containern mit SageMakeraus.