Tentukan pipa

1. Buka konsol Studio dengan mengikuti petunjuk diLuncurkan Amazon SageMaker Studio.

2. Di panel navigasi kiri, pilih Pipelines.

3. Pilih Buat.

4. Pilih Kosong.

5. Di bilah sisi kiri, pilih Proses data dan seret ke kanvas.

6. Di kanvas, pilih langkah Proses data yang Anda tambahkan.

7. Untuk menambahkan kumpulan data input, pilih Tambah di bawah Data (input) di bilah sisi kanan dan pilih kumpulan data.

8. Untuk menambahkan lokasi untuk menyimpan kumpulan data keluaran, pilih Tambah di bawah Data (output) di bilah sisi kanan dan arahkan ke tujuan.

9. Lengkapi bidang yang tersisa di sidebar kanan. Untuk informasi tentang bidang di tab ini, lihat sagemaker.workflow.steps. ProcessingStep.

1. Di sidebar kiri, pilih Train model dan seret ke kanvas.

2. Di kanvas, pilih langkah model Kereta yang Anda tambahkan.

3. Untuk menambahkan kumpulan data input, pilih Tambah di bawah Data (input) di bilah sisi kanan dan pilih kumpulan data.

4. Untuk memilih lokasi untuk menyimpan artefak model Anda, masukkan Amazon URI S3 di bidang Lokasi (URIS3), atau pilih Jelajahi S3 untuk menavigasi ke lokasi tujuan.

5. Lengkapi bidang yang tersisa di sidebar kanan. Untuk informasi tentang bidang di tab ini, lihat sagemaker.workflow.steps. TrainingStep.

6. Klik dan seret kursor dari langkah Proses data yang Anda tambahkan di bagian sebelumnya ke langkah Model kereta untuk membuat tepi yang menghubungkan dua langkah.

1. Di bilah sisi kiri, pilih Daftar model dan seret ke kanvas.

2. Di kanvas, pilih langkah model Daftar yang Anda tambahkan.

3. Untuk memilih model yang akan didaftarkan, pilih Tambah di bawah Model (input).

4. Pilih Buat grup model untuk menambahkan model Anda ke grup model baru.

5. Lengkapi bidang yang tersisa di sidebar kanan. Untuk informasi tentang bidang di tab ini, lihat sagemaker.workflow.step_collections. RegisterModel.

6. Klik dan seret kursor dari langkah model Kereta yang Anda tambahkan di bagian sebelumnya ke langkah model Daftar untuk membuat tepi yang menghubungkan dua langkah.

1. Di sidebar kiri, pilih Deploy model (endpoint) dan seret ke kanvas.

2. Di kanvas, pilih langkah Deploy model (endpoint) yang Anda tambahkan.

3. Untuk memilih model yang akan digunakan, pilih Tambah di bawah Model (input).

4. Pilih tombol Create endpoint radio untuk membuat endpoint baru.

5. Masukkan Nama dan Deskripsi untuk titik akhir Anda.

6. Klik dan seret kursor dari langkah model Register yang Anda tambahkan di bagian sebelumnya ke langkah Deploy model (endpoint) untuk membuat tepi yang menghubungkan dua langkah.

7. Lengkapi bidang yang tersisa di sidebar kanan.

Untuk mengunduh dataset

1. Unduh kumpulan data ke bucket Amazon S3 default akun Anda.

   !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. Unduh kumpulan data kedua untuk transformasi batch setelah model Anda dibuat.

   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)

Untuk membuat langkah pemrosesan

1. Buat direktori untuk skrip pemrosesan.

   !mkdir -p abalone

2. Buat file di /abalone direktori bernama preprocessing.py dengan konten berikut. Skrip preprocessing ini diteruskan ke langkah pemrosesan untuk berjalan pada data input. Langkah pelatihan kemudian menggunakan fitur dan label pelatihan yang telah diproses sebelumnya untuk melatih model. Langkah evaluasi menggunakan model terlatih dan fitur dan label uji yang telah diproses sebelumnya untuk mengevaluasi model. Skrip digunakan scikit-learn untuk melakukan hal berikut:

   • Isi data sex kategoris yang hilang dan kodekan sehingga cocok untuk pelatihan.

   • Skala dan normalkan semua bidang numerik kecuali untuk rings dansex.

   • Pisahkan data menjadi kumpulan data pelatihan, pengujian, dan validasi.

   %%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. Buat instance SKLearnProcessor untuk diteruskan ke langkah pemrosesan.

   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. Buat langkah pemrosesan. Langkah ini mengambilSKLearnProcessor, saluran input dan output, dan preprocessing.py skrip yang Anda buat. Ini sangat mirip dengan run metode instance prosesor di SageMaker PythonSDK. input_dataParameter yang diteruskan ProcessingStep adalah data input dari langkah itu sendiri. Data input ini digunakan oleh instance prosesor saat berjalan.

   Perhatikan"train,"validation, dan saluran "test" bernama yang ditentukan dalam konfigurasi output untuk pekerjaan pemrosesan. Langkah Properties seperti ini dapat digunakan dalam langkah-langkah berikutnya dan menyelesaikan nilai runtime mereka saat runtime.

   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
   )

Untuk menentukan langkah pelatihan

1. Tentukan jalur model tempat Anda ingin menyimpan model dari pelatihan.

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

2. Konfigurasikan estimator untuk XGBoost algoritme dan kumpulan data input. Jenis instance pelatihan diteruskan ke estimator. Skrip pelatihan yang khas:

   • memuat data dari saluran input

   • mengonfigurasi pelatihan dengan hyperparameters

   • melatih model

   • menyimpan model model_dir agar dapat di-host nanti

   SageMaker mengunggah model ke Amazon S3 dalam bentuk model.tar.gz a di akhir pekerjaan pelatihan.

   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. Buat TrainingStep menggunakan instance estimator dan properti dari. ProcessingStep Lewati S3Uri saluran "train" dan "validation" output keTrainingStep. 

   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
   )

Untuk menentukan langkah pemrosesan untuk evaluasi model

1. Buat file di /abalone direktori bernamaevaluation.py. Skrip ini digunakan dalam langkah pemrosesan untuk melakukan evaluasi model. Dibutuhkan model terlatih dan dataset pengujian sebagai input, kemudian menghasilkan JSON file yang berisi metrik evaluasi klasifikasi.

   %%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. Buat sebuah instance dari sebuah ScriptProcessor yang digunakan untuk membuat sebuahProcessingStep.

   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. Buat ProcessingStep menggunakan instance prosesor, saluran input dan output, dan evaluation.py skrip. Lulus:

   • S3ModelArtifactsproperti dari langkah step_train pelatihan

   • S3Urisaluran "test" output dari langkah step_process pemrosesan

   Ini sangat mirip dengan run metode instance prosesor di SageMaker PythonSDK. 

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

Untuk menentukan transformasi CreateModelStep untuk batch

1. Buat SageMaker model. Lewati S3ModelArtifacts properti dari langkah step_train pelatihan.

   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. Tentukan input model untuk SageMaker model Anda.

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

3. Buat contoh CreateModelStep menggunakan CreateModelInput dan SageMaker model yang Anda tentukan.

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

Untuk menentukan TransformStep untuk melakukan transformasi batch

1. Buat instance transformator dengan jenis instans komputasi yang sesuai, jumlah instans, dan bucket Amazon URI S3 keluaran yang diinginkan. Lewati ModelName properti dari step_create_model CreateModel langkah.

   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. Buat TransformStep menggunakan instance transformator yang Anda tentukan dan parameter batch_data pipeline.

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

Untuk menentukan RegisterModel langkah untuk membuat paket model

• Buat RegisterModel langkah menggunakan instance estimator yang Anda gunakan untuk langkah pelatihan. Lewati S3ModelArtifacts properti dari langkah step_train pelatihan dan tentukan aModelPackageGroup. Pipelines menciptakan ini ModelPackageGroup untuk Anda.

  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
  )

Untuk menentukan langkah kondisi untuk memverifikasi akurasi model

1. Tentukan ConditionLessThanOrEqualTo kondisi menggunakan nilai akurasi yang ditemukan dalam output dari langkah pemrosesan evaluasi model,step_eval. Dapatkan output ini menggunakan file properti yang Anda indeks dalam langkah pemrosesan dan masing-masing JSONPath nilai kesalahan kuadrat rata-rata,. "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. Membangun a. ConditionStep Lulus ConditionEquals kondisi, lalu atur pendaftaran paket model dan langkah transformasi batch sebagai langkah selanjutnya jika kondisi berlalu.

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

Untuk membuat pipa

1. Tentukan hal berikut untuk pipeline Anda:name,parameters, dansteps. Nama harus unik dalam (account, region) pasangan.

   catatan

   Sebuah langkah hanya dapat muncul sekali di daftar langkah pipeline atau daftar langkah jika/lain dari langkah kondisi. Itu tidak bisa muncul di keduanya.

   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. (Opsional) Periksa definisi JSON pipa untuk memastikan bahwa itu terbentuk dengan baik.

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