Definieren Sie eine Pipeline

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

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

3. Wählen Sie Create (Erstellen) aus.

4. Wählen Sie Leer aus.

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

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 die Option Modell trainieren und ziehen Sie es auf die Leinwand.

2. Wählen Sie auf der Leinwand 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 einen Amazon S3 URI in das Feld Standort (S3URI) ein, oder wählen Sie Browse S3, 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 Schritt Prozessdaten, 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 Tabs finden Sie unter sagemaker.workflow.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 Arbeitsfläche.

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 Ihren Endpunkt ein.

6. Klicken und ziehen Sie den Cursor vom Schritt Modell registrieren, den Sie im vorherigen Abschnitt hinzugefügt haben, zum 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 Trainingsfunktionen und Labels, um ein Modell zu trainieren. Im Bewertungsschritt werden das trainierte Modell und die vorverarbeiteten Testmerkmale und Bezeichnungen verwendet, 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 Prozessorinstanz in SageMaker Python sehr ähnlichSDK. 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 zur Laufzeit in ihre Laufzeitwerte aufgelöst werden.

   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 damit es später gehostet werden kann

   SageMaker lädt das Modell am Ende der Schulung in Form eines Jobs 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 TrainingStep unter Verwendung des Estimators eine Instance und die Eigenschaften von. ProcessingStep Übergeben Sie den S3Uri von "train" und den "validation" Ausgangskanal an denTrainingStep. 

   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 Sie ProcessingStep mithilfe des Prozessors eine Instanz, die Eingabe- und Ausgabekanäle und das evaluation.py Skript. Weitergeben:

   • die S3ModelArtifacts Immobilie aus dem step_train Trainingsschritt

   • die S3Uri des "test" Ausgangskanals des step_process Verarbeitungsschritts

   Dies ist der run Methode einer Prozessorinstanz in SageMaker Python sehr ähnlichSDK. 

   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 eine CreateModelStep Batch-Transformation zu definieren

1. Erstellen Sie ein SageMaker 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 Modell.

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

3. Erstellen Sie Ihre CreateModelStep unter Verwendung der CreateModelInput von Ihnen definierten SageMaker Modellinstanz.

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

Um eine TransformStep Batch-Transformation zu definieren

1. Erstellen Sie eine Transformer-Instance mit dem entsprechenden Compute-Instance-Typ, der Instance-Anzahl und dem gewünschten Amazon S3 S3-Ausgabe-BucketURI. Ü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 das 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 quadratischen Fehlerwert JSONPath des mittleren Quadrats. "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())