Definieren einer Pipeline

1. Öffnen Sie die Studio-Konsole, indem Sie den Anweisungen unter folgenStarten von Amazon SageMaker Studio.

2. Wählen Sie im linken Navigationsbereich die Option Pipelines aus.

3. Wählen Sie Erstellen aus.

4. Wählen Sie Leer.

5. Wählen Sie in der linken Seitenleiste die Option Daten verarbeiten und ziehen Sie sie auf die Arbeitsfläche.

6. Wählen Sie auf der Arbeitsfläche den Schritt „Daten verarbeiten“ aus, den Sie hinzugefügt haben.

7. Um einen Eingabedatensatz hinzuzufügen, wählen Sie in der rechten Seitenleiste unter Daten (Eingabe) die Option Hinzufügen und wählen Sie einen Datensatz aus.

8. Um einen Speicherort für Ausgabe-Datasets hinzuzufügen, wählen Sie in der rechten Seitenleiste unter Daten (Ausgabe) die Option Hinzufügen und navigieren Sie zum Ziel.

9. Füllen Sie die verbleibenden Felder in der rechten Seitenleiste aus. Informationen zu den Feldern in diesen Tabs finden Sie unter SageMaker.Workflow.Steps.ProcessingStep.

1. Wählen Sie in der linken Seitenleiste Train Model und ziehen Sie es auf die Leinwand.

2. Wählen Sie auf der Arbeitsfläche den Schritt Train Model aus, den Sie hinzugefügt haben.

3. Um einen Eingabedatensatz hinzuzufügen, wählen Sie in der rechten Seitenleiste unter Daten (Eingabe) die Option Hinzufügen und wählen Sie einen Datensatz aus.

4. Um einen Speicherort für Ihre Modellartefakte auszuwählen, geben Sie eine Amazon S3 S3-URI in das Feld Standort (S3-URI) ein oder wählen Sie S3 durchsuchen, um zum Zielort zu navigieren.

5. Füllen Sie die verbleibenden Felder in der rechten Seitenleiste aus. Informationen zu den Feldern in diesen Tabs finden Sie unter SageMaker.Workflow.Steps.TrainingStep.

6. Klicken und ziehen Sie den Cursor aus dem Prozessdatenschritt, den Sie im vorherigen Abschnitt hinzugefügt haben, in den Schritt Modell trainieren, um eine Kante zu erzeugen, die die beiden Schritte verbindet.

1. Wählen Sie in der linken Seitenleiste Modell registrieren und ziehen Sie es auf die Arbeitsfläche.

2. Wählen Sie auf der Arbeitsfläche den Schritt Modell registrieren aus, den Sie hinzugefügt haben.

3. Um ein Modell für die Registrierung auszuwählen, wählen Sie unter Modell (Eingabe) die Option Hinzufügen aus.

4. Wählen Sie Modellgruppe erstellen, um Ihr Modell zu einer neuen Modellgruppe hinzuzufügen.

5. Füllen Sie die verbleibenden Felder in der rechten Seitenleiste aus. Informationen zu den Feldern in diesen Registerkarten finden Sie unter SageMaker.step_collections.RegisterModel.

6. Klicken und ziehen Sie den Cursor aus dem Schritt Modell trainieren, den Sie im vorherigen Abschnitt hinzugefügt haben, in den Schritt Modell registrieren, um eine Kante zu erzeugen, die die beiden Schritte verbindet.

1. Wählen Sie in der linken Seitenleiste Modell bereitstellen (Endpunkt) aus und ziehen Sie es auf die Leinwand.

2. Wählen Sie auf der Arbeitsfläche den Schritt Modell bereitstellen (Endpunkt) aus, den Sie hinzugefügt haben.

3. Um ein Modell für die Bereitstellung auszuwählen, wählen Sie unter Modell (Eingabe) die Option Hinzufügen aus.

4. Wählen Sie das Optionsfeld Endpunkt erstellen, um einen neuen Endpunkt zu erstellen.

5. Geben Sie einen Namen und eine Beschreibung für den Endpunkt ein.

6. Klicken und ziehen Sie den Cursor aus dem Schritt Modell registrieren, den Sie im vorherigen Abschnitt hinzugefügt haben, in den Schritt Modell bereitstellen (Endpunkt), um eine Kante zu erstellen, die die beiden Schritte verbindet.

7. Füllen Sie die verbleibenden Felder in der rechten Seitenleiste aus.

Zum Herunterladen des Datensatzes

1. Laden Sie den Datensatz in den standardmäßigen Amazon-S3-Bucket Ihres Kontos herunter.

   !mkdir -p data
   local_path = "data/abalone-dataset.csv"
   
   s3 = boto3.resource("s3")
   s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file(
       "dataset/abalone-dataset.csv",
       local_path
   )
   
   base_uri = f"s3://{default_bucket}/abalone"
   input_data_uri = sagemaker.s3.S3Uploader.upload(
       local_path=local_path, 
       desired_s3_uri=base_uri,
   )
   print(input_data_uri)

2. Laden Sie nach der Erstellung Ihres Modells einen zweiten Datensatz für die Batch-Transformation herunter.

   local_path = "data/abalone-dataset-batch.csv"
   
   s3 = boto3.resource("s3")
   s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file(
       "dataset/abalone-dataset-batch",
       local_path
   )
   
   base_uri = f"s3://{default_bucket}/abalone"
   batch_data_uri = sagemaker.s3.S3Uploader.upload(
       local_path=local_path, 
       desired_s3_uri=base_uri,
   )
   print(batch_data_uri)

Um einen Verarbeitungsschritt zu erstellen

1. Erstellen Sie ein Verzeichnis für das Verarbeitungsskript.

   !mkdir -p abalone

2. Erstellen Sie im Verzeichnis /abalone eine Datei namens preprocessing.py mit folgendem Inhalt. Dieses Vorverarbeitungsskript wird an den Verarbeitungsschritt zur Ausführung der Eingabedaten übergeben. Der Trainingsschritt verwendet dann die vorverarbeiteten Trainingsmerkmale und Labels, um ein Modell zu trainieren. Der Bewertungsschritt verwendet das trainierte Modell und die vorverarbeiteten Testmerkmale und Labels, um das Modell zu evaluieren. Das Skript verwendet scikit-learn für die folgenden Aufgaben:

   • Füllen Sie fehlende sex kategoriale Daten aus und codieren Sie sie so, dass sie für das Training geeignet sind.

   • Skalieren und normalisieren Sie alle numerischen Felder außer rings und sex.

   • Teilen Sie die Daten in Trainings-, Test- und Validierungsdatensätze auf.

   %%writefile abalone/preprocessing.py
   import argparse
   import os
   import requests
   import tempfile
   import numpy as np
   import pandas as pd
   
   
   from sklearn.compose import ColumnTransformer
   from sklearn.impute import SimpleImputer
   from sklearn.pipeline import Pipeline
   from sklearn.preprocessing import StandardScaler, OneHotEncoder
   
   
   # Because this is a headerless CSV file, specify the column names here.
   feature_columns_names = [
       "sex",
       "length",
       "diameter",
       "height",
       "whole_weight",
       "shucked_weight",
       "viscera_weight",
       "shell_weight",
   ]
   label_column = "rings"
   
   feature_columns_dtype = {
       "sex": str,
       "length": np.float64,
       "diameter": np.float64,
       "height": np.float64,
       "whole_weight": np.float64,
       "shucked_weight": np.float64,
       "viscera_weight": np.float64,
       "shell_weight": np.float64
   }
   label_column_dtype = {"rings": np.float64}
   
   
   def merge_two_dicts(x, y):
       z = x.copy()
       z.update(y)
       return z
   
   
   if __name__ == "__main__":
       base_dir = "/opt/ml/processing"
   
       df = pd.read_csv(
           f"{base_dir}/input/abalone-dataset.csv",
           header=None, 
           names=feature_columns_names + [label_column],
           dtype=merge_two_dicts(feature_columns_dtype, label_column_dtype)
       )
       numeric_features = list(feature_columns_names)
       numeric_features.remove("sex")
       numeric_transformer = Pipeline(
           steps=[
               ("imputer", SimpleImputer(strategy="median")),
               ("scaler", StandardScaler())
           ]
       )
   
       categorical_features = ["sex"]
       categorical_transformer = Pipeline(
           steps=[
               ("imputer", SimpleImputer(strategy="constant", fill_value="missing")),
               ("onehot", OneHotEncoder(handle_unknown="ignore"))
           ]
       )
   
       preprocess = ColumnTransformer(
           transformers=[
               ("num", numeric_transformer, numeric_features),
               ("cat", categorical_transformer, categorical_features)
           ]
       )
       
       y = df.pop("rings")
       X_pre = preprocess.fit_transform(df)
       y_pre = y.to_numpy().reshape(len(y), 1)
       
       X = np.concatenate((y_pre, X_pre), axis=1)
       
       np.random.shuffle(X)
       train, validation, test = np.split(X, [int(.7*len(X)), int(.85*len(X))])
   
       
       pd.DataFrame(train).to_csv(f"{base_dir}/train/train.csv", header=False, index=False)
       pd.DataFrame(validation).to_csv(f"{base_dir}/validation/validation.csv", header=False, index=False)
       pd.DataFrame(test).to_csv(f"{base_dir}/test/test.csv", header=False, index=False)

3. Erstellen Sie eine Instance von SKLearnProcessor, die an den Verarbeitungsschritt übergeben werden soll.

   from sagemaker.sklearn.processing import SKLearnProcessor
   
   
   framework_version = "0.23-1"
   
   sklearn_processor = SKLearnProcessor(
       framework_version=framework_version,
       instance_type="ml.m5.xlarge",
       instance_count=processing_instance_count,
       base_job_name="sklearn-abalone-process",
       sagemaker_session=pipeline_session,
       role=role,
   )

4. Erstellen Sie einen Verarbeitungsschritt. In diesem Schritt werden die SKLearnProcessor Eingabe- und Ausgabekanäle sowie das von Ihnen erstellte preprocessing.py Skript berücksichtigt. Dies ist der run-Methode einer Prozessor-Instance im SageMaker AI Python SDK sehr ähnlich. Der Parameter input_data, der an ProcessingStep übergeben wird, sind die Eingabedaten des Schrittes selbst. Diese Eingabedaten werden von der Prozessor-Instance verwendet, wenn sie ausgeführt wird.

   Beachten Sie die in der Ausgabekonfiguration für den Verarbeitungsauftrag angegebenen "train, "validation, und "test" benannten Kanäle. Schritte Properties wie diese können in nachfolgenden Schritten verwendet werden und werden bei der Laufzeit in ihre Laufzeitwerte umgewandelt.

   from sagemaker.processing import ProcessingInput, ProcessingOutput
   from sagemaker.workflow.steps import ProcessingStep
      
   
   processor_args = sklearn_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="abalone/preprocessing.py",
   ) 
   
   step_process = ProcessingStep(
       name="AbaloneProcess",
       step_args=processor_args
   )

Um einen Trainingsschritt zu definieren

1. Geben Sie den Modellpfad an, in dem Sie die Modelle aus dem Training speichern möchten.

   model_path = f"s3://{default_bucket}/AbaloneTrain"

2. Konfigurieren Sie einen Schätzer für den XGBoost-Algorithmus und den Eingabedatensatz. Der Typ der Trainings-Instance wird an den Schätzer übergeben. Ein typisches Trainingsskript:

   • lädt Daten aus den Eingangskanälen

   • konfiguriert das Training mit Hyperparametern

   • trainiert ein Modell

   • speichert ein Modell unter, model_dir sodass es später gehostet werden kann

   SageMaker AI lädt das Modell am Ende des Trainingsauftrags in Form einer model.tar.gz auf Amazon S3 hoch.

   from sagemaker.estimator import Estimator
   
   
   image_uri = sagemaker.image_uris.retrieve(
       framework="xgboost",
       region=region,
       version="1.0-1",
       py_version="py3",
       instance_type="ml.m5.xlarge"
   )
   xgb_train = Estimator(
       image_uri=image_uri,
       instance_type="ml.m5.xlarge",
       instance_count=1,
       output_path=model_path,
       sagemaker_session=pipeline_session,
       role=role,
   )
   xgb_train.set_hyperparameters(
       objective="reg:linear",
       num_round=50,
       max_depth=5,
       eta=0.2,
       gamma=4,
       min_child_weight=6,
       subsample=0.7,
       silent=0
   )

3. Erstellen Sie einen TrainingStep unter Verwendung der Schätzfunktion-Instance und der Eigenschaften des ProcessingStep. Leiten Sie den S3Uri des – "train"- und "validation"-Ausgangskanals an den TrainingStep weiter. 

   from sagemaker.inputs import TrainingInput
   from sagemaker.workflow.steps import TrainingStep
   
   
   train_args = xgb_train.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="AbaloneTrain",
       step_args = train_args
   )

Um einen Verarbeitungsschritt für die Modellevaluierung zu definieren

1. Erstellen Sie im Verzeichnis /abalone eine Datei mit dem Namen evaluation.py. Dieses Skript wird in einem Verarbeitungsschritt zur Durchführung der Modellevaluierung verwendet. Es verwendet ein trainiertes Modell und den Testdatensatz als Eingabe und erstellt dann eine JSON-Datei mit Bewertungsmetriken für die Klassifizierung.

   %%writefile abalone/evaluation.py
   import json
   import pathlib
   import pickle
   import tarfile
   import joblib
   import numpy as np
   import pandas as pd
   import xgboost
   
   
   from sklearn.metrics import mean_squared_error
   
   
   if __name__ == "__main__":
       model_path = f"/opt/ml/processing/model/model.tar.gz"
       with tarfile.open(model_path) as tar:
           tar.extractall(path=".")
       
       model = pickle.load(open("xgboost-model", "rb"))
   
       test_path = "/opt/ml/processing/test/test.csv"
       df = pd.read_csv(test_path, header=None)
       
       y_test = df.iloc[:, 0].to_numpy()
       df.drop(df.columns[0], axis=1, inplace=True)
       
       X_test = xgboost.DMatrix(df.values)
       
       predictions = model.predict(X_test)
   
       mse = mean_squared_error(y_test, predictions)
       std = np.std(y_test - predictions)
       report_dict = {
           "regression_metrics": {
               "mse": {
                   "value": mse,
                   "standard_deviation": std
               },
           },
       }
   
       output_dir = "/opt/ml/processing/evaluation"
       pathlib.Path(output_dir).mkdir(parents=True, exist_ok=True)
       
       evaluation_path = f"{output_dir}/evaluation.json"
       with open(evaluation_path, "w") as f:
           f.write(json.dumps(report_dict))

2. Erstellen Sie eine Instance von ScriptProcessor, die verwendet wird, um eine ProcessingStep zu erstellen.

   from sagemaker.processing import ScriptProcessor
   
   
   script_eval = ScriptProcessor(
       image_uri=image_uri,
       command=["python3"],
       instance_type="ml.m5.xlarge",
       instance_count=1,
       base_job_name="script-abalone-eval",
       sagemaker_session=pipeline_session,
       role=role,
   )

3. Erstellen eines ProcessingStep Auftrags mithilfe des Prozessors, der Eingabe- und evaluation.py Ausgabekanäle. Übergeben:

   • die S3ModelArtifacts-Eigenschaft aus dem step_train-Trainingsschritt

   • der S3Uri des "test" Ausgangskanals des step_process Verarbeitungsschritts

   Dies ist der run-Methode einer Prozessor-Instance im SageMaker AI Python SDK sehr ähnlich. 

   from sagemaker.workflow.properties import PropertyFile
   
   
   evaluation_report = PropertyFile(
       name="EvaluationReport",
       output_name="evaluation",
       path="evaluation.json"
   )
   
   eval_args = script_eval.run(
           inputs=[
           ProcessingInput(
               source=step_train.properties.ModelArtifacts.S3ModelArtifacts,
               destination="/opt/ml/processing/model"
           ),
           ProcessingInput(
               source=step_process.properties.ProcessingOutputConfig.Outputs[
                   "test"
               ].S3Output.S3Uri,
               destination="/opt/ml/processing/test"
           )
       ],
       outputs=[
           ProcessingOutput(output_name="evaluation", source="/opt/ml/processing/evaluation"),
       ],
       code="abalone/evaluation.py",
   )
   
   step_eval = ProcessingStep(
       name="AbaloneEval",
       step_args=eval_args,
       property_files=[evaluation_report],
   )

Um einen CreateModelStep für die Batch-Transformation zu definieren

1. Erstellen Sie ein SageMaker-AI-Modell. Übergeben Sie die S3ModelArtifacts Eigenschaft aus dem step_train Trainingsschritt.

   from sagemaker.model import Model
   
   
   model = Model(
       image_uri=image_uri,
       model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
       sagemaker_session=pipeline_session,
       role=role,
   )

2. Definieren Sie die Modelleingabe für Ihr SageMaker-AI-Modell.

   from sagemaker.inputs import CreateModelInput
   
   
   inputs = CreateModelInput(
       instance_type="ml.m5.large",
       accelerator_type="ml.eia1.medium",
   )

3. Erstellen Sie Ihren CreateModelStep unter Verwendung der CreateModelInput und von Ihnen definierten Modell-Instance von SageMaker AI.

   from sagemaker.workflow.steps import CreateModelStep
   
   
   step_create_model = CreateModelStep(
       name="AbaloneCreateModel",
       model=model,
       inputs=inputs,
   )

Um einen TransformStep zur Durchführung einer Batch-Transformation zu definieren

1. Erstellen Sie eine Transformer-Instance mit dem entsprechenden Compute-Instance-Typ, der Instance-Anzahl und der gewünschten Amazon S3-Output-Bucket-URI. Übergeben Sie die ModelName Eigenschaft aus dem step_create_model CreateModel Schritt.

   from sagemaker.transformer import Transformer
   
   
   transformer = Transformer(
       model_name=step_create_model.properties.ModelName,
       instance_type="ml.m5.xlarge",
       instance_count=1,
       output_path=f"s3://{default_bucket}/AbaloneTransform"
   )

2. Erstellen Sie eine TransformStep mit der Transformer-Instance, die Sie definiert haben, und dem batch_data Pipeline-Parameter.

   from sagemaker.inputs import TransformInput
   from sagemaker.workflow.steps import TransformStep
   
   
   step_transform = TransformStep(
       name="AbaloneTransform",
       transformer=transformer,
       inputs=TransformInput(data=batch_data)
   )

Um einen RegisterModel-Schritt zur Erstellung eines Modellpakets zu definieren

• Konstruieren Sie einen RegisterModel Schritt mit der Estimator-Instance, die Sie für den Trainingsschritt verwendet haben. Übergeben Sie die S3ModelArtifacts Eigenschaft aus dem step_train Trainingsschritt und geben Sie ModelPackageGroup an. Pipelines erstellt diese ModelPackageGroup für Sie.

  from sagemaker.model_metrics import MetricsSource, ModelMetrics 
  from sagemaker.workflow.step_collections import RegisterModel
  
  
  model_metrics = ModelMetrics(
      model_statistics=MetricsSource(
          s3_uri="{}/evaluation.json".format(
              step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
          ),
          content_type="application/json"
      )
  )
  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
  )

Um einen Bedingungsschritt zur Überprüfung der Modellgenauigkeit zu definieren

1. Definieren Sie eine ConditionLessThanOrEqualTo Bedingung anhand des Genauigkeitswerts, der in der Ausgabe des Verarbeitungsschritts der Modellbewertung ermittelt wurde, step_eval. Verwenden Sie für diese Ausgabe die Eigenschaftendatei, die Sie im Verarbeitungsschritt indexiert haben, und den entsprechenden JSONPath des mittleren quadratischen Fehlerwerts, "mse".

   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
   )

2. Konstruieren Sie ein ConditionStep. Übergeben Sie die ConditionEquals Bedingung und legen Sie dann die Schritte zur Registrierung des Modellpakets und zur Batch-Transformation als nächste Schritte fest, wenn die Bedingung erfüllt ist.

   step_cond = ConditionStep(
       name="AbaloneMSECond",
       conditions=[cond_lte],
       if_steps=[step_register, step_create_model, step_transform],
       else_steps=[], 
   )

So erstellen Sie eine Pipeline

1. Definieren Sie Folgendes für Ihre Pipeline: name, parameters, undsteps. Die Namen müssen innerhalb eines (account, region)-Paares eindeutig sein.

   Anmerkung

   Ein Schritt kann entweder in der Schrittliste der Pipeline oder in den if/else-Schrittlisten des Bedingungsschritts nur einmal vorkommen. Er kann nicht in beiden vorkommen.

   from sagemaker.workflow.pipeline import Pipeline
   
   
   pipeline_name = f"AbalonePipeline"
   pipeline = Pipeline(
       name=pipeline_name,
       parameters=[
           processing_instance_count,
           model_approval_status,
           input_data,
           batch_data,
       ],
       steps=[step_process, step_train, step_eval, step_cond],
   )

2. (Optional) Untersuchen Sie die JSON-Pipeline-Definition, um sicherzustellen, dass sie korrekt formatiert ist.

   import json
   
   json.loads(pipeline.definition())