Aggiunta di una fase

Nell’interfaccia utente con trascinamento di Pipelines, puoi utilizzare una fase Esecuzione di codice per eseguire il codice come fase della pipeline. Puoi caricare una funzione, uno script o un notebook Python da eseguire come parte della tua pipeline. Devi utilizzare questa fase se intendi orchestrare un processo di ML personalizzato che sfrutta le funzionalità avanzate di SageMaker AI o altri servizi AWS.

La fase Esecuzione di codice carica i file nel bucket Amazon S3 predefinito per Amazon SageMaker AI. Questo bucket potrebbe non disporre del set di autorizzazioni Condivisione delle risorse multiorigine (CORS) richiesto. Per ulteriori informazioni sulla configurazione delle autorizzazioni CORS, consulta Requisito CORS per i dati delle immagini di input.

La fase Esecuzione di codice utilizza un job di addestramento di Amazon SageMaker per eseguire il codice. Verifica che il tuo ruolo IAM disponga delle autorizzazioni API sagemaker:DescribeTrainingJob e sagemaker:CreateTrainingJob. Per ulteriori informazioni su tutte le autorizzazioni richieste per Amazon SageMaker AI e su come configurarle, consulta Documentazione di riferimento sulle autorizzazioni delle API di Amazon SageMaker AI: azioni, autorizzazioni e risorse.

Per aggiungere una fase Esecuzione di codice a una pipeline utilizzando Pipeline Designer, procedi come descritto di seguito:

Utilizza una fase di elaborazione per creare un processo di elaborazione per l'elaborazione dei dati. Per ulteriori informazioni sull'elaborazione dei processi, consulta Elaborazione di dati e valutazione di modelli.

Si utilizza una fase di addestramento per creare un processo di addestramento per addestrare un modello. Per ulteriori informazioni sui job di addestramento, consulta Addestramento di un modello con Amazon SageMaker AI.

Una fase di addestramento richiede uno strumento di valutazione e input di dati di addestramento e convalida.

Si utilizza una fase di ottimizzazione per creare un processo di ottimizzazione degli iperparametri, noto anche come ottimizzazione degli iperparametri (HPO). Un processo di ottimizzazione degli iperparametri esegue più job di addestramento, ognuno dei quali produce una versione del modello. Per ulteriori informazioni sull'ottimizzazione degli iperparametri, consulta Ottimizzazione automatica dei modelli con SageMaker AI.

Il processo di ottimizzazione è associato all’esperimento di SageMaker AI per la pipeline, con i job di addestramento creati come prove. Per ulteriori informazioni, consulta Integrazione di Esperimenti.

Una fase di ottimizzazione richiede un HyperparameterTuner e input di addestramento. È possibile addestrare nuovamente processi di ottimizzazione precedenti specificando il parametro warm_start_config del HyperparameterTuner. Per ulteriori informazioni sull'ottimizzazione degli iperparametri e sull'avvio a caldo, consulta Eseguire un processo di ottimizzazione degli iperparametri con avvio a caldo.

Si utilizza il metodo get_top_model_s3_uri della classe sagemaker.workflow.steps.TuningStep per ottenere l'artefatto del modello da una delle versioni del modello con le migliori prestazioni. Per un notebook che mostra come utilizzare una fase di ottimizzazione in una pipeline SageMaker AI, consulta sagemaker-pipelines-tuning-step.ipynb.

L'esempio seguente mostra come creare una definizione di TuningStep.

from sagemaker.workflow.pipeline_context import PipelineSession

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

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

Ottenere la versione migliore del modello

Nell'esempio seguente viene illustrato come ottenere la versione migliore del modello dal processo di ottimizzazione mediante il metodo get_top_model_s3_uri. Al massimo, sono disponibili le 50 versioni con le migliori prestazioni, classificate in base a HyperParameterTuningJobObjective. L'argomento top_k è un indice delle versioni, dove top_k=0 indica la versione con le migliori prestazioni e top_k=49 quella con le prestazioni peggiori.

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

Per ulteriori informazioni sui requisiti della fase di ottimizzazione, consulta la documentazione sagemaker.workflow.steps.TuningStep.

Il fine-tuning consente di addestrare un modello di fondazione preaddestrato da Amazon SageMaker JumpStart su un nuovo set di dati. Questo processo, noto anche come trasferimento dell'apprendimento, può produrre modelli accurati con set di dati più piccoli e tempi di addestramento ridotti. Quando esegui il fine-tuning di un modello, puoi utilizzare il set di dati predefinito o scegliere i tuoi dati. Per ulteriori informazioni sul fine-tuning di un modello di fondazione con JumpStart, consulta Ottimizzare un modello.

La fase di fine-tuning utilizza un job di addestramento di Amazon SageMaker per personalizzare il modello. Verifica che il tuo ruolo IAM disponga delle autorizzazioni API sagemaker:DescribeTrainingJob e sagemaker:CreateTrainingJob necessarie per eseguire il processo di fine-tuning nella tua pipeline. Per ulteriori informazioni sulle autorizzazioni necessarie per Amazon SageMaker AI e su come configurarle, consulta Documentazione di riferimento sulle autorizzazioni delle API di Amazon SageMaker AI: azioni, autorizzazioni e risorse.

Per aggiungere una fase Fine-tuning di un modello alla tua pipeline utilizzando l’editor con trascinamento, segui questa procedura:

Utilizza l'API AutoML per creare un processo AutoML per addestrare automaticamente un modello. Per ulteriori informazioni sui processi AutoML, consulta Automatizzazione dello sviluppo di modelli con il pilota automatico Amazon SageMaker.

L'esempio seguente mostra come creare una definizione utilizzando AutoMLStep.

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

pipeline_session = PipelineSession()

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

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

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

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

Ottenere la versione migliore del modello

La fase AutoML addestra automaticamente diversi candidati modello. È possibile ottenere il modello con la metrica obiettivo ottimale dal processo AutoML utilizzando il metodo get_best_auto_ml_model come segue. È inoltre necessario utilizzare un role IAM per accedere agli artefatti del modello.

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

Per ulteriori informazioni, consulta la fase AutoML in SageMaker Python SDK.

Utilizza ModelStep per creare o registrare un modello SageMaker AI. Per ulteriori informazioni sui requisiti di ModelStep, consulta la documentazione sagemaker.workflow.model_step.ModelStep.

Creazione di un modello

Puoi utilizzare ModelStep per creare un modello SageMaker AI. ModelStep richiede artefatti del modello e informazioni sul tipo di istanza SageMaker AI che è necessario utilizzare per creare il modello. Per ulteriori informazioni sui modelli SageMaker AI, consulta Addestramento di un modello con Amazon SageMaker AI.

L'esempio seguente mostra come creare una definizione di 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"),
)

Registrazione di un modello

Puoi utilizzare ModelStep per registrare sagemaker.model.Model o sagemaker.pipeline.PipelineModel in Amazon SageMaker Model Registry. Un PipelineModel rappresenta una pipeline di inferenza, che è un modello composto da una sequenza lineare di container che elaborano richieste di inferenza. Per ulteriori informazioni su come eseguire la registrazione di un modello, consulta Implementazione della registrazione del modello con il registro dei modelli.

L'esempio seguente mostra come creare una ModelStep per registrare un PipelineModel.

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

La fase Creazione di un modello serve a creare un modello SageMaker AI. Per ulteriori informazioni sui modelli SageMaker AI, consulta Addestramento di un modello con Amazon SageMaker.

Una fase Creazione di un modello richiede artefatti del modello e informazioni sul tipo di istanza SageMaker AI che è necessario utilizzare per creare il modello. Gli esempi seguenti mostrano come creare una definizione della fase Creazione di un modello. Per ulteriori informazioni sui requisiti della fase Creazione di un modello, consulta la documentazione sagemaker.workflow.steps.CreateModelStep.

La fase Registrazione di un modello serve a registrare un modello in SageMaker Model Registry.

In Pipeline Designer, utilizza la fase Implementazione di un modello (endpoint) per implementare il modello in un endpoint. Puoi creare un nuovo endpoint o utilizzarne uno esistente. L'inferenza in tempo reale è ideale per carichi di lavoro di inferenza in cui sono previsti requisiti in tempo reale, interattivi e a bassa latenza. È possibile implementare il modello sui servizi di Hosting SageMaker AI e ottenere un endpoint che può essere utilizzato per l’inferenza. Questi endpoint sono completamente gestiti e supportano il dimensionamento automatico. Per saperne di più sull’inferenza in tempo reale in SageMaker AI, consulta Inferenza in tempo reale.

Prima di aggiungere una fase Implementazione di un modello alla pipeline, verifica che il tuo ruolo IAM abbia le autorizzazioni seguenti:

Per ulteriori informazioni su tutte le autorizzazioni necessarie per SageMaker AI e su come configurarle, consulta Documentazione di riferimento sulle autorizzazioni delle API di Amazon SageMaker AI: azioni, autorizzazioni e risorse.

Per aggiungere una fase di implementazione del modello alla pipeline nell’editor con trascinamento, procedi come descritto di seguito:

Si utilizza una fase di trasformazione in modo che la trasformazione in batch esegua inferenza su un intero set di dati. Per ulteriori informazioni sulla trasformazione in batch, consulta Trasformazioni in batch con pipeline di inferenza.

Una fase di trasformazione richiede un trasformatore e i dati su cui eseguire la trasformazione in batch. L’esempio seguente mostra come creare una definizione della fase Trasformazione. Per ulteriori informazioni sui requisiti della fase Trasformazione, consulta la documentazione sagemaker.workflow.steps.TransformStep.

Si utilizza una fase condizionale per valutare la condizione delle proprietà della fase per valutare quale azione deve essere intrapresa successivamente nella pipeline.

Una fase di condizione richiede:

Utilizza una fase Callback per aggiungere al tuo flusso di lavoro altri processi e servizi AWS che non sono forniti direttamente da Amazon SageMaker Pipelines. Quando viene eseguita una fase di Callback, si verifica la seguente procedura:

Per ulteriori informazioni sui requisiti della fase Callback, consulta la documentazione sagemaker.workflow.callback_step.CallbackStep. Per una soluzione completa, consulta Estensione di SageMaker Pipelines per includere fasi personalizzate utilizzando fasi di callback.

L’esempio seguente illustra un’implementazione della procedura precedente.

from sagemaker.workflow.callback_step import CallbackStep

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

callback_handler_code = '
    import boto3
    import json

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

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

            # Custom processing

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

Comportamento di arresto

Un processo di pipeline non si arresta durante l'esecuzione di una fase di Callback.

Quando chiami StopPipelineExecution su un processo della pipeline con una fase Callback in esecuzione, Pipelines invia un messaggio Amazon SQS alla coda SQS. Il corpo del messaggio SQS contiene un campo Stato, impostato su Stopping. Nell'esempio seguente viene mostrato il corpo di un messaggio 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"
}

È opportuno aggiungere una logica al consumer del messaggio Amazon SQS affinché intraprenda le eventuali azioni necessarie (ad esempio, la pulizia delle risorse) al momento della ricezione del messaggio. Quindi, aggiungi una chiamata a SendPipelineExecutionStepSuccess o SendPipelineExecutionStepFailure.

Solo quando Pipelines riceve una di queste chiamate arresta il processo della pipeline.

Si utilizza una fase Lambda per eseguire una funzione AWS Lambda. Puoi eseguire una funzione Lambda esistente, oppure SageMaker AI può creare ed eseguire una nuova funzione Lambda. Se scegli di utilizzare una funzione Lambda esistente, questa deve trovarsi nella stessa Regione AWS della pipeline SageMaker AI. Per un notebook che mostra come utilizzare una fase Lambda in una pipeline SageMaker AI, consulta sagemaker-pipelines-lambda-step.ipynb.

SageMaker AI fornisce la classe sagemaker.lambda_helper.Lambda per creare, aggiornare, invocare ed eliminare le funzioni Lambda. Lambda ha la seguente firma.

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

La classe sagemaker.workflow.lambda_step.LambdaStep ha un argomento lambda_func di tipo Lambda. Per richiamare una funzione Lambda esistente, l'unico requisito è fornire il nome della risorsa Amazon (ARN) della funzione a function_arn. Se non fornisci un valore per function_arn, devi specificare handler e uno dei seguenti parametri:

L'esempio seguente mostra come creare una definizione di fase Lambda che richiami una funzione Lambda esistente.

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

L'esempio seguente mostra come creare una definizione di fase Lambda che crei e richiami una funzione Lambda utilizzando uno script di funzione 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 e output

Se la funzione Lambda ha input o output, anche questi devono essere definiti nella fase Lambda.

Quando si definisce la fase Lambda, inputs deve essere un dizionario di coppie chiave-valore. Ogni valore del dizionario inputs deve essere di tipo primitivo (stringa, numero intero o float). Gli oggetti nidificati non sono supportati. Se non viene definito, il valore inputs predefinito è None.

Il valore outputs deve essere un elenco di chiavi. Queste chiavi si riferiscono a un dizionario definito nell'output della funzione Lambda. Come per inputs, queste chiavi devono essere di tipo primitivo e gli oggetti nidificati non sono supportati.

Comportamento di timeout e arresto

La classe Lambda ha un argomento timeout che specifica il tempo massimo di esecuzione della funzione Lambda. Il valore predefinito è 120 secondi con un valore massimo di 10 minuti. Se la funzione Lambda è in esecuzione quando viene raggiunto il timeout, la fase Lambda ha esito negativo; tuttavia, la funzione Lambda continua a essere in esecuzione.

Un processo di pipeline non può essere arrestato mentre è in esecuzione una fase Lambda perché la funzione Lambda richiamata dalla fase Lambda non può essere arrestata. Se provi ad arrestare il processo mentre la funzione Lambda è in esecuzione, la pipeline attende che la funzione venga completata o raggiunga il timeout, a seconda dell’evento che si verifica per primo. A questo punto, il processo si arresta. Se la funzione Lambda termina, lo stato del processo di pipeline è Stopped. Se viene raggiunto il timeout, lo stato del processo di pipeline è Failed.

Si può utilizzare la fase ClarifyCheck per eseguire controlli della deviazione dalla baseline rispetto alle baseline precedenti per l'analisi dei bias e la spiegabilità del modello. È quindi possibile generare e registrare le proprie baseline con il metodo model.register() e passare l'output di tale metodo a Fase del modello utilizzando step_args. Le baseline per il controllo della deriva possono essere utilizzate da Amazon SageMaker Model Monitor per gli endpoint del modello. Di conseguenza, non è necessario eseguire separatamente un suggerimento baseline.

La fase ClarifyCheck può anche estrarre le baseline per il controllo della deviazione dal registro dei modelli. La fase ClarifyCheck utilizza il container predefinito SageMaker Clarify. Questo container offre una gamma di funzionalità di monitoraggio dei modelli, tra cui suggerimenti e convalida dei vincoli rispetto a una determinata baseline. Per ulteriori informazioni, consulta Container predefiniti di SageMaker Clarify.

Configurazione della fase ClarifyCheck

È possibile configurare la fase ClarifyCheck per eseguire solo uno dei seguenti tipi di controllo ogni volta che viene utilizzata in una pipeline.

Per farlo, imposta il parametro clarify_check_config con uno dei valori del tipo di controllo seguenti:

La fase ClarifyCheck avvia un processo di elaborazione che esegue il container predefinito SageMaker AI Clarify e richiede configurazioni dedicate per il controllo e il processo di elaborazione. ClarifyCheckConfig e CheckJobConfig sono funzioni di supporto per queste configurazioni. Queste funzioni di supporto sono in linea con il metodo di calcolo con cui il processo di elaborazione di SageMaker Clarify verifica i bias del modello, i bias dei dati o la spiegabilità dei modelli. Per ulteriori informazioni, consulta Esegui i processi di elaborazione SageMaker Clarify per l'analisi e la spiegabilità dei bias.

Controllo dei comportamenti della fase per il controllo della deviazione

La fase ClarifyCheck richiede i seguenti due flag booleani per controllarne il comportamento:

Per ulteriori informazioni, consulta Calcolo della baseline, rilevamento della deriva e ciclo di vita con le fasi ClarifyCheck e QualityCheck in Amazon SageMaker Pipelines.

Utilizzo delle baseline

Puoi anche specificare model_package_group_name per individuare la baseline esistente. Quindi, la fase ClarifyCheck estrae DriftCheckBaselines nell’ultimo pacchetto di modelli approvato nel gruppo di pacchetti di modelli.

In alternativa, è possibile fornire una baseline precedente tramite il parametro supplied_baseline_constraints. Se si specificano sia model_package_group_name che supplied_baseline_constraints, la fase ClarifyCheck utilizza la baseline specificata dal parametro supplied_baseline_constraints.

Per ulteriori informazioni sull’utilizzo dei requisiti della fase ClarifyCheck, consulta sagemaker.workflow.steps.ClarifyCheckStep in SDK per Python di Amazon SageMaker AI. Per un notebook Amazon SageMaker Studio Classic che mostra come utilizzare la fase ClarifyCheck in Pipelines, consulta 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"
)

Utilizza la fase QualityCheck per eseguire suggerimenti baseline e controlli della deriva rispetto a una baseline precedente per verificare la qualità dei dati o del modello in una pipeline. Puoi quindi generare e registrare le tue baseline con il metodo model.register() e passare l’output di tale metodo a Fase del modello utilizzando step_args.

Model Monitor può usare queste baseline per il controllo della deviazione per gli endpoint del modello in modo da non dover eseguire un suggerimento di baseline separatamente. La fase QualityCheck può anche estrarre le baseline per il controllo della deviazione dal registro dei modelli. La fase QualityCheck sfrutta il container predefinito Amazon SageMaker AI Model Monitor. Questo container offre una gamma di funzionalità di monitoraggio dei modelli, tra cui la generazione di statistiche e i suggerimenti e la convalida dei vincoli rispetto a una determinata baseline. Per ulteriori informazioni, consulta Container predefinito Amazon SageMaker Model Monitor.

Configurazione della fase QualityCheck

Puoi configurare la fase QualityCheck per eseguire solo uno dei seguenti tipi di controllo ogni volta che viene utilizzato in una pipeline.

È possibile eseguire questa operazione impostando il parametro quality_check_config con uno dei valori di tipo di controllo seguenti:

La fase QualityCheck avvia un processo di elaborazione che esegue il container esistente di Model Monitor e richiede configurazioni dedicate per il controllo e il processo di elaborazione. QualityCheckConfig e CheckJobConfig sono funzioni di supporto per queste configurazioni. Queste funzioni di supporto sono in linea con il modo in cui Model Monitor crea una baseline per il monitoraggio della qualità del modello o dei dati. Per ulteriori informazioni sui suggerimenti di baseline di Model Monitor, consulta Creazione di una linea di base e Creazione di una baseline per la qualità del modello.

Controllo dei comportamenti della fase per il controllo della deviazione

La fase QualityCheck richiede i seguenti due flag booleani per controllarne il comportamento:

Per ulteriori informazioni, consulta Calcolo della baseline, rilevamento della deriva e ciclo di vita con le fasi ClarifyCheck e QualityCheck in Amazon SageMaker Pipelines.

Utilizzo delle baseline

Puoi specificare una baseline precedente direttamente tramite i parametri supplied_baseline_statistics e supplied_baseline_constraints. Puoi anche specificare model_package_group_name per far sì che la fase QualityCheck estragga DriftCheckBaselines nell’ultimo pacchetto di modelli approvato nel gruppo di pacchetti di modelli.

Quando specifichi quanto segue, la fase QualityCheck utilizza la baseline indicata da supplied_baseline_constraints e supplied_baseline_statistics sul tipo di controllo della fase QualityCheck.

Per ulteriori informazioni sull’utilizzo dei requisiti della fase QualityCheck, consulta sagemaker.workflow.steps.QualityCheckStep in SDK per Python di Amazon SageMaker AI. Per un notebook Amazon SageMaker Studio Classic che mostra come utilizzare la fase QualityCheck in Pipelines, consulta 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"
)

Utilizza la fase EMR di Amazon SageMaker Pipelines per:

Per ulteriori informazioni su Amazon EMR, consulta Nozioni di base su Amazon EMR.

La fase EMR richiede che EMRStepConfig includa la posizione del file JAR utilizzato dal cluster Amazon EMR e tutti gli argomenti da passare. Fornisci anche l’ID del cluster Amazon EMR se intendi eseguire la fase su un cluster EMR in esecuzione. Puoi anche passare la configurazione del cluster per eseguire la fase EMR su un cluster creato, gestito e terminato al posto tuo. Le sezioni seguenti includono esempi e collegamenti a notebook di esempio che illustrano entrambi i metodi.

Avvio di un nuovo processo su un cluster Amazon EMR in esecuzione

Per avviare un nuovo processo su un cluster Amazon EMR in esecuzione, passa l’ID cluster come stringa all’argomento cluster_id di EMRStep. L'esempio seguente illustra questa procedura.

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

Per un notebook di esempio con indicazioni dettagliate per l’esempio completo, consulta Fase EMR di Pipelines con cluster EMR in esecuzione.

Avvio di un nuovo processo su un nuovo cluster Amazon EMR

Per avviare un nuovo processo su un nuovo cluster creato per te da EMRStep, fornisci la configurazione del cluster come dizionario, che deve avere la stessa struttura di una richiesta RunJobFlow. Tuttavia, non includere i seguenti campi nella configurazione del cluster:

Tutti gli altri argomenti RunJobFlow sono disponibili per l'uso nella configurazione del cluster. Per i dettagli sulla sintassi della richiesta, consulta RunJobFlow.

L’esempio seguente passa una configurazione del cluster a una definizione della fase EMR. Alla fase viene richiesto di avviare un nuovo processo su un nuovo cluster EMR. La configurazione del cluster EMR in questo esempio include le specifiche per i nodi primari e principali del cluster EMR. Per ulteriori informazioni sui tipi di nodo Amazon EMR, consulta Informazioni sui tipi di nodi: nodi primari, principali e attività.

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
)

Per un notebook di esempio con indicazioni dettagliate per l’esempio completo, consulta Fase EMR di Pipelines con gestione del ciclo di vita del cluster.

Utilizza NotebookJobStep per eseguire il Processo del notebook SageMaker in modo non interattivo come fase della pipeline. Se crei la pipeline nell’interfaccia utente con trascinamento di Pipelines, utilizza Fase Esecuzione di codice per eseguire il notebook. Per ulteriori informazioni su Processi del notebook SageMaker, consulta SageMaker Notebook Jobs.

NotebookJobStep richiede almeno un notebook di input, un URI dell’immagine e un nome del kernel. Per ulteriori informazioni sui requisiti per la fase Processo del notebook e su altri parametri che puoi impostare per personalizzare la fase, consulta sagemaker.workflow.steps.NotebookJobStep.

L’esempio seguente utilizza un numero minimo di argomenti per definire 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
)

La fase della pipeline NotebookJobStep viene considerata come un processo del notebook SageMaker. Di conseguenza, monitora lo stato di esecuzione nella dashboard dei processi del notebook nell’interfaccia utente di Studio Classic includendo tag specifici con argomento tags. Per ulteriori informazioni sui tag da utilizzare, consulta Visualizzazione dei processi del notebook nella dashboard dell’interfaccia utente di Studio.

Inoltre, se pianifichi il processo del notebook utilizzando SageMaker Python SDK, puoi specificare solo determinate immagini per eseguire il processo del notebook. Per ulteriori informazioni, consulta Vincoli di immagine per i processi del notebook SageMaker AI Python SDK.

Utilizza la fase Errore per arrestare un’esecuzione di Amazon SageMaker Pipelines quando una condizione o uno stato desiderato non viene raggiunto. La fase Errore consente inoltre di inserire un messaggio di errore personalizzato, che indica la causa dell’errore di esecuzione della pipeline.

Limitazioni all’utilizzo della fase Errore

Puoi creare il messaggio di errore della fase Errore sotto forma di stringa di testo statica. In alternativa, puoi anche utilizzare i Parametri della pipeline, un’operazione Esegui join o altre proprietà della fase per creare un messaggio di errore più informativo, se utilizzi l’SDK.