Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
# Preprocessing dan Postprocessing
Anda dapat menggunakan skrip Python preprocessing dan postprocessing khusus untuk mengubah input ke monitor model Anda atau memperluas kode setelah pemantauan berhasil dijalankan. Unggah skrip ini ke Amazon S3 dan referensikan saat membuat monitor model Anda.
Contoh berikut menunjukkan bagaimana Anda dapat menyesuaikan jadwal pemantauan dengan skrip preprocessing dan postprocessing. Ganti *user placeholder text* dengan informasi Anda sendiri.
```
import boto3, os
from sagemaker import get_execution_role, Session
from sagemaker.model_monitor import CronExpressionGenerator, DefaultModelMonitor
# Upload pre and postprocessor scripts
session = Session()
bucket = boto3.Session().resource("s3").Bucket(session.default_bucket())
prefix = "demo-sagemaker-model-monitor"
pre_processor_script = bucket.Object(os.path.join(prefix, "preprocessor.py")).upload_file("preprocessor.py")
post_processor_script = bucket.Object(os.path.join(prefix, "postprocessor.py")).upload_file("postprocessor.py")
# Get execution role
role = get_execution_role() # can be an empty string
# Instance type
instance_type = "instance-type"
# instance_type = "ml.m5.xlarge" # Example
# Create a monitoring schedule with pre and postprocessing
my_default_monitor = DefaultModelMonitor(
role=role,
instance_count=1,
instance_type=instance_type,
volume_size_in_gb=20,
max_runtime_in_seconds=3600,
)
s3_report_path = "s3://{}/{}".format(bucket, "reports")
monitor_schedule_name = "monitor-schedule-name"
endpoint_name = "endpoint-name"
my_default_monitor.create_monitoring_schedule(
post_analytics_processor_script=post_processor_script,
record_preprocessor_script=pre_processor_script,
monitor_schedule_name=monitor_schedule_name,
# use endpoint_input for real-time endpoint
endpoint_input=endpoint_name,
# or use batch_transform_input for batch transform jobs
# batch_transform_input=batch_transform_name,
output_s3_uri=s3_report_path,
statistics=my_default_monitor.baseline_statistics(),
constraints=my_default_monitor.suggested_constraints(),
schedule_cron_expression=CronExpressionGenerator.hourly(),
enable_cloudwatch_metrics=True,
)
```
**Topics**
+ [Skrip Preprocessing](#model-monitor-pre-processing-script)
+ [Pengambilan Sampel Kustom](#model-monitor-pre-processing-custom-sampling)
+ [Skrip Pasca Pemrosesan](#model-monitor-post-processing-script)
## Skrip Preprocessing
Gunakan skrip preprocessing saat Anda perlu mengubah input ke monitor model Anda.
Sebagai contoh, misalkan output dari model Anda adalah array`[1.0, 2.1]`. Wadah Amazon SageMaker Model Monitor hanya berfungsi dengan struktur JSON tabular atau pipih, seperti. `{“prediction0”: 1.0, “prediction1” : 2.1}` Anda dapat menggunakan skrip preprocessing seperti berikut ini untuk mengubah array menjadi struktur JSON yang benar.
```
def preprocess_handler(inference_record):
input_data = inference_record.endpoint_input.data
output_data = inference_record.endpoint_output.data.rstrip("\n")
data = output_data + "," + input_data
return { str(i).zfill(20) : d for i, d in enumerate(data.split(",")) }
```
Dalam contoh lain, misalkan model Anda memiliki fitur opsional dan Anda gunakan `-1` untuk menunjukkan bahwa fitur opsional memiliki nilai yang hilang. Jika Anda memiliki monitor kualitas data, Anda mungkin ingin menghapus `-1` dari array nilai input sehingga tidak termasuk dalam perhitungan metrik monitor. Anda dapat menggunakan skrip seperti berikut ini untuk menghapus nilai-nilai tersebut.
```
def preprocess_handler(inference_record):
input_data = inference_record.endpoint_input.data
return {i : None if x == -1 else x for i, x in enumerate(input_data.split(","))}
```
Script preprocessing Anda menerima `inference_record` sebagai satu-satunya masukan. Cuplikan kode berikut menunjukkan contoh file. `inference_record`
```
{
"captureData": {
"endpointInput": {
"observedContentType": "text/csv",
"mode": "INPUT",
"data": "132,25,113.2,96,269.9,107,,0,0,0,0,0,0,1,0,1,0,0,1",
"encoding": "CSV"
},
"endpointOutput": {
"observedContentType": "text/csv; charset=utf-8",
"mode": "OUTPUT",
"data": "0.01076381653547287",
"encoding": "CSV"
}
},
"eventMetadata": {
"eventId": "feca1ab1-8025-47e3-8f6a-99e3fdd7b8d9",
"inferenceTime": "2019-11-20T23:33:12Z"
},
"eventVersion": "0"
}
```
Cuplikan kode berikut menunjukkan struktur kelas penuh untuk sebuah. `inference_record`
```
KEY_EVENT_METADATA = "eventMetadata"
KEY_EVENT_METADATA_EVENT_ID = "eventId"
KEY_EVENT_METADATA_EVENT_TIME = "inferenceTime"
KEY_EVENT_METADATA_CUSTOM_ATTR = "customAttributes"
KEY_EVENTDATA_ENCODING = "encoding"
KEY_EVENTDATA_DATA = "data"
KEY_GROUND_TRUTH_DATA = "groundTruthData"
KEY_EVENTDATA = "captureData"
KEY_EVENTDATA_ENDPOINT_INPUT = "endpointInput"
KEY_EVENTDATA_ENDPOINT_OUTPUT = "endpointOutput"
KEY_EVENTDATA_BATCH_OUTPUT = "batchTransformOutput"
KEY_EVENTDATA_OBSERVED_CONTENT_TYPE = "observedContentType"
KEY_EVENTDATA_MODE = "mode"
KEY_EVENT_VERSION = "eventVersion"
class EventConfig:
def __init__(self, endpoint, variant, start_time, end_time):
self.endpoint = endpoint
self.variant = variant
self.start_time = start_time
self.end_time = end_time
class EventMetadata:
def __init__(self, event_metadata_dict):
self.event_id = event_metadata_dict.get(KEY_EVENT_METADATA_EVENT_ID, None)
self.event_time = event_metadata_dict.get(KEY_EVENT_METADATA_EVENT_TIME, None)
self.custom_attribute = event_metadata_dict.get(KEY_EVENT_METADATA_CUSTOM_ATTR, None)
class EventData:
def __init__(self, data_dict):
self.encoding = data_dict.get(KEY_EVENTDATA_ENCODING, None)
self.data = data_dict.get(KEY_EVENTDATA_DATA, None)
self.observedContentType = data_dict.get(KEY_EVENTDATA_OBSERVED_CONTENT_TYPE, None)
self.mode = data_dict.get(KEY_EVENTDATA_MODE, None)
def as_dict(self):
ret = {
KEY_EVENTDATA_ENCODING: self.encoding,
KEY_EVENTDATA_DATA: self.data,
KEY_EVENTDATA_OBSERVED_CONTENT_TYPE: self.observedContentType,
}
return ret
class CapturedData:
def __init__(self, event_dict):
self.event_metadata = None
self.endpoint_input = None
self.endpoint_output = None
self.batch_transform_output = None
self.ground_truth = None
self.event_version = None
self.event_dict = event_dict
self._event_dict_postprocessed = False
if KEY_EVENT_METADATA in event_dict:
self.event_metadata = EventMetadata(event_dict[KEY_EVENT_METADATA])
if KEY_EVENTDATA in event_dict:
if KEY_EVENTDATA_ENDPOINT_INPUT in event_dict[KEY_EVENTDATA]:
self.endpoint_input = EventData(event_dict[KEY_EVENTDATA][KEY_EVENTDATA_ENDPOINT_INPUT])
if KEY_EVENTDATA_ENDPOINT_OUTPUT in event_dict[KEY_EVENTDATA]:
self.endpoint_output = EventData(event_dict[KEY_EVENTDATA][KEY_EVENTDATA_ENDPOINT_OUTPUT])
if KEY_EVENTDATA_BATCH_OUTPUT in event_dict[KEY_EVENTDATA]:
self.batch_transform_output = EventData(event_dict[KEY_EVENTDATA][KEY_EVENTDATA_BATCH_OUTPUT])
if KEY_GROUND_TRUTH_DATA in event_dict:
self.ground_truth = EventData(event_dict[KEY_GROUND_TRUTH_DATA])
if KEY_EVENT_VERSION in event_dict:
self.event_version = event_dict[KEY_EVENT_VERSION]
def as_dict(self):
if self._event_dict_postprocessed is True:
return self.event_dict
if KEY_EVENTDATA in self.event_dict:
if KEY_EVENTDATA_ENDPOINT_INPUT in self.event_dict[KEY_EVENTDATA]:
self.event_dict[KEY_EVENTDATA][KEY_EVENTDATA_ENDPOINT_INPUT] = self.endpoint_input.as_dict()
if KEY_EVENTDATA_ENDPOINT_OUTPUT in self.event_dict[KEY_EVENTDATA]:
self.event_dict[KEY_EVENTDATA][
KEY_EVENTDATA_ENDPOINT_OUTPUT
] = self.endpoint_output.as_dict()
if KEY_EVENTDATA_BATCH_OUTPUT in self.event_dict[KEY_EVENTDATA]:
self.event_dict[KEY_EVENTDATA][KEY_EVENTDATA_BATCH_OUTPUT] = self.batch_transform_output.as_dict()
self._event_dict_postprocessed = True
return self.event_dict
def __str__(self):
return str(self.as_dict())
```
## Pengambilan Sampel Kustom
Anda juga dapat menerapkan strategi pengambilan sampel khusus dalam skrip preprocessing Anda. Untuk melakukan ini, konfigurasikan wadah pra-bangun pihak pertama Model Monitor untuk mengabaikan persentase catatan sesuai dengan laju pengambilan sampel yang Anda tentukan. Dalam contoh berikut, handler mengambil sampel 10 persen dari catatan dengan mengembalikan catatan dalam 10 persen panggilan handler dan daftar kosong sebaliknya.
```
import random
def preprocess_handler(inference_record):
# we set up a sampling rate of 0.1
if random.random() > 0.1:
# return an empty list
return []
input_data = inference_record.endpoint_input.data
return {i : None if x == -1 else x for i, x in enumerate(input_data.split(","))}
```
### Logging kustom untuk skrip preprocessing
Jika skrip preprocessing Anda mengembalikan kesalahan, periksa pesan pengecualian yang masuk CloudWatch ke debug. Anda dapat mengakses logger CloudWatch melalui `preprocess_handler` antarmuka. Anda dapat mencatat informasi apa pun yang Anda butuhkan dari skrip Anda CloudWatch. Ini dapat berguna saat men-debug skrip preprocessing Anda. Contoh berikut menunjukkan bagaimana Anda dapat menggunakan `preprocess_handler` antarmuka untuk log ke CloudWatch
```
def preprocess_handler(inference_record, logger):
logger.info(f"I'm a processing record: {inference_record}")
logger.debug(f"I'm debugging a processing record: {inference_record}")
logger.warning(f"I'm processing record with missing value: {inference_record}")
logger.error(f"I'm a processing record with bad value: {inference_record}")
return inference_record
```
## Skrip Pasca Pemrosesan
Gunakan skrip postprocessing saat Anda ingin memperpanjang kode setelah pemantauan yang berhasil dijalankan.
```
def postprocess_handler():
print("Hello from post-proc script!")
```