本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。 # 進階主題 下列各節包含更進階的任務，說明如何使用預先處理和後製處理指令碼自訂監控、如何建置您自己的容器，以及如何使用 CloudFormation 建立監控排程。 **Topics** + [自訂監控排程](model-monitor-custom-monitoring-schedules.md) + [使用 CloudFormation 自訂資源建立即時端點的監控排程](model-monitor-cloudformation-monitoring-schedules.md) # 自訂監控排程 除了使用內建的監控機制之外，您還可以使用預處理和後處理指令碼，或使用或建置您自己的容器，以建立您自己的自訂監控排程和程序。 **Topics** + [預處理和後處理](model-monitor-pre-and-post-processing.md) + [使用 Amazon SageMaker Model Monitor 支援您自己的容器](model-monitor-byoc-containers.md) # 預處理和後處理 您可以使用自訂的預先處理和後製處理 Python 指令碼，將輸入轉換至模型監控，或在成功執行監控後擴充程式碼。將這些指令碼上傳到 Amazon S3，並在建立模型監控時參照它們。 下列範例顯示如何使用預先處理和後製處理指令碼來自訂監控排程。將*使用者預留位置文字*取代為自己的資訊。 ``` 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** + [預處理指令碼](#model-monitor-pre-processing-script) + [自訂取樣](#model-monitor-pre-processing-custom-sampling) + [後處理指令碼](#model-monitor-post-processing-script) ## 預處理指令碼 當您需要將輸入轉換為模型監控時，請使用預先處理指令碼。 例如，假設模型的輸出是陣列 `[1.0, 2.1]`。Amazon SageMaker Model Monitor 容器僅適用於表格式或扁平化 JSON 結構，例如 `{“prediction0”: 1.0, “prediction1” : 2.1}`。您可以使用如下的預先處理指令碼，將陣列轉換為正確的 JSON 結構。 ``` 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(",")) } ``` 在另一個範例中，假設您的模型具有可選功能，並且您使用 `-1` 來表示可選功能有缺少值。如果您有資料品質監控，您可能想要將 `-1` 從輸入值陣列中移除，這樣它就不會包含在監控的指標計算中。您可以使用如下所示的指令碼來移除這些值。 ``` 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(","))} ``` 您的預先處理指令碼會接收 `inference_record` 為其唯一的輸入。下列程式碼片段說明 `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" } ``` 下列程式碼片段顯示 `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()) ``` ## 自訂取樣 您也可以在預先處理指令碼中套用自訂取樣策略。若要這麼做，請設定模型監控的第一方預先建置容器，根據您指定的取樣率忽略某個百分比的記錄。在下列範例中，處理常式會取樣 10% 的記錄，傳回 10% 的處理常式呼叫的記錄，其餘為空白清單。 ``` 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(","))} ``` ### 預先處理指令碼的自訂記錄 如果您的預先處理指令碼傳回錯誤，請檢查記錄至 CloudWatch 的例外狀況訊息以進行偵錯。您可以在 CloudWatch 上透過 `preprocess_handler` 介面來存取記錄器。您可以將指令碼中所需的任何資訊記錄至 CloudWatch。這在對預先處理指令碼進行偵錯時非常有用。以下範例顯示如何使用 `preprocess_handler` 介面記錄至 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 ``` ## 後處理指令碼 如果您想要在成功執行監控之後擴充程式碼，請使用後製處理指令碼。 ``` def postprocess_handler(): print("Hello from post-proc script!") ``` # 使用 Amazon SageMaker Model Monitor 支援您自己的容器 Amazon SageMaker Model Monitor 提供預先建置的容器，能夠分析從表格式資料集的端點或批次轉換工作擷取的資料。如果您想使用自有容器，模型監控提供擴充點供您利用。 在幕後，當您建立 `MonitoringSchedule` 時，模型監控最終會啟動處理工作。因此，容器需要知道 [如何建置您自己的處理容器 (進階案例)](build-your-own-processing-container.md) 主題中所記載的處理工作合約。請注意，模型監控會根據排程為您啟動處理工作。調用時，模型監控會為您設定額外的環境變數，以針對排程監控的該特定執行，讓容器有足夠的情境來處理資料。如需容器輸入的其他資訊，請參閱[容器合約輸入](model-monitor-byoc-contract-inputs.md)。 在容器中，您現在可以使用上述環境變數/情境，在自訂程式碼中針對目前時段分析資料集。完成此分析後，您可以選擇發出報告以上傳至 S3 儲存貯體。預先建置容器產生的報告記載於 [容器合約輸出](model-monitor-byoc-contract-outputs.md)。如果您想要在 SageMaker Studio 中啟用報告視覺化，則應該遵循相同的格式。您也可以選擇發出完全自訂的報告。 您也可以依照[適用於使用自有容器的 CloudWatch 指標](model-monitor-byoc-cloudwatch.md)中的指示，從容器發出 CloudWatch 指標。 **Topics** + [容器合約輸入](model-monitor-byoc-contract-inputs.md) + [容器合約輸出](model-monitor-byoc-contract-outputs.md) + [適用於使用自有容器的 CloudWatch 指標](model-monitor-byoc-cloudwatch.md) # 容器合約輸入 Amazon SageMaker Model Monitor 平台根據指定的排程，調用您的容器程式碼。如果您選擇撰寫自己的容器程式碼，下列環境變數可使用。在此情況下，您可以分析目前資料集或評估限制條件 (如果選擇如此)，並發出指標 (如果適用)。 即時端點和批次轉換工作的可用環境變數相同，`dataset_format` 變數除外。如果您使用的是即時端點，`dataset_format` 變數會支援下列選項： ``` {\"sagemakerCaptureJson\": {\"captureIndexNames\": [\"endpointInput\",\"endpointOutput\"]}} ``` 如果您使用的是批次轉換工作，則 `dataset_format` 支援下列選項： ``` {\"csv\": {\"header\": [\"true\",\"false\"]}} ``` ``` {\"json\": {\"line\": [\"true\",\"false\"]}} ``` ``` {\"parquet\": {}} ``` 下列程式碼範例顯示可用於容器程式碼的完整環境變數集 (並使用即時端點的 `dataset_format` 格式)。 ``` "Environment": { "dataset_format": "{\"sagemakerCaptureJson\": {\"captureIndexNames\": [\"endpointInput\",\"endpointOutput\"]}}", "dataset_source": "/opt/ml/processing/endpointdata", "end_time": "2019-12-01T16: 20: 00Z", "output_path": "/opt/ml/processing/resultdata", "publish_cloudwatch_metrics": "Disabled", "sagemaker_endpoint_name": "endpoint-name", "sagemaker_monitoring_schedule_name": "schedule-name", "start_time": "2019-12-01T15: 20: 00Z" } ``` Parameters | 參數名稱 | Description | | --- | --- | | dataset\$1format | 對於從 `Endpoint` 支援的 `MonitoringSchedule` 開始的工作，此為 `sageMakerCaptureJson` 且擷取索引為 `endpointInput`、`endpointOutput` 或兩者兼有。對於批次轉換工作，這會指定資料格式，無論是 CSV、JSON 或 Parquet。 | | dataset\$1source | 如果您使用的是即時端點，可使用與 `start_time` 和 `end_time` 指定的監控期間對應的資料所在的本機路徑。在此路徑中，資料位於 ` /{endpoint-name}/{variant-name}/yyyy/mm/dd/hh` 中。 我們有時會下載超過開始和結束時間所指定的資料量。容器程式碼會依需要來剖析資料。 | | output\$1path | 寫入輸出報告和其他檔案的本機路徑。請在 `CreateMonitoringSchedule` 請求中以 `MonitoringOutputConfig.MonitoringOutput[0].LocalPath` 指定此參數。它會上傳到 `MonitoringOutputConfig.MonitoringOutput[0].S3Uri` 中指定的 `S3Uri` 路徑。 | | publish\$1cloudwatch\$1metrics | 對於由 `CreateMonitoringSchedule` 啟動的工作，此參數設定為 `Enabled`。容器可以選擇在 `[filepath]` 寫入 Amazon CloudWatch 輸出檔案。 | | sagemaker\$1endpoint\$1name | 如果您使用的是即時端點，則為啟動此排程工作的 `Endpoint` 名稱。 | | sagemaker\$1monitoring\$1schedule\$1name | 啟動此工作的 `MonitoringSchedule` 名稱。 | | \$1sagemaker\$1endpoint\$1datacapture\$1prefix\$1 | 如果您使用的是即時端點，則為 `Endpoint` 的 `DataCaptureConfig` 參數中指定的前置詞。如果容器需要直接存取超過 SageMaker AI 在 `dataset_source` 路徑上已下載的資料量，則可以使用此參數。 | | start\$1time, end\$1time | 此分析執行的時間範圍。例如，對於排定在 05:00 (UTC) 執行的工作和 2020/2/20 執行的工作，`start_time` 為 2020-02-19T06:00:00Z，`end_time` 為 2020-02-20T05:00:00Z | | baseline\$1constraints: | ` BaselineConfig.ConstraintResource.S3Uri` 中指定的基準限制條件檔案的本機路徑。這只有當 `CreateMonitoringSchedule` 請求中指定此參數時才能使用。 | | baseline\$1statistics | `BaselineConfig.StatisticsResource.S3Uri` 中指定的基準統計資料檔案的本機路徑。這只有當 `CreateMonitoringSchedule` 請求中指定此參數時才能使用。 | # 容器合約輸出 容器可以分析 `*dataset_source*` 路徑中可用的資料，並將報告寫入 `*output_path*.` 中的路徑。容器程式碼可以撰寫符合您需求的任何報告。 如果您使用下列結構和合約，SageMaker AI 在視覺化和 API 上會特別處理某些輸出檔案。這只適用於表格式資料集。 表格式資料集的輸出檔案 | 檔案名稱 | Description | | --- | --- | | statistics.json | 針對所分析資料集的每個特徵，此檔案會有單欄式統計資料。在下一節可查看此檔案的結構描述。 | | constraints.json | 針對所觀察的特徵，此檔案會有限制條件。在下一節可查看此檔案的結構描述。 | | constraints\$1violations.json | 與 `baseline_constaints` 和 `baseline_statistics` 路徑中指定的基準統計資料和限制條件檔案相比較之後，此檔案中預期會有目前這組資料中發現的違規清單。 | 此外，如果 `publish_cloudwatch_metrics` 值為 `"Enabled"`，容器程式碼可以在此位置發出 Amazon CloudWatch 指標：`/opt/ml/output/metrics/cloudwatch`。下列各節描述這些檔案的結構描述。 **Topics** + [統計資料的結構描述 (statistics.json 檔案)](model-monitor-byoc-statistics.md) + [限制條件的結構描述 (constraints.json 檔案)](model-monitor-byoc-constraints.md) # 統計資料的結構描述 (statistics.json 檔案) 針對基準和擷取的資料，`statistics.json` 檔案中定義的結構描述指定要計算的統計參數。另外還設定儲存貯體供 [KLL](https://datasketches.apache.org/docs/KLL/KLLSketch.html) 使用 (非常簡潔的分位數草圖，具有延遲壓縮配置)。 ``` { "version": 0, # dataset level stats "dataset": { "item_count": number }, # feature level stats "features": [ { "name": "feature-name", "inferred_type": "Fractional" | "Integral", "numerical_statistics": { "common": { "num_present": number, "num_missing": number }, "mean": number, "sum": number, "std_dev": number, "min": number, "max": number, "distribution": { "kll": { "buckets": [ { "lower_bound": number, "upper_bound": number, "count": number } ], "sketch": { "parameters": { "c": number, "k": number }, "data": [ [ num, num, num, num ], [ num, num ][ num, num ] ] }#sketch }#KLL }#distribution }#num_stats }, { "name": "feature-name", "inferred_type": "String", "string_statistics": { "common": { "num_present": number, "num_missing": number }, "distinct_count": number, "distribution": { "categorical": { "buckets": [ { "value": "string", "count": number } ] } } }, #provision for custom stats } ] } ``` **備註** 在稍後的視覺化變更中，SageMaker AI 可辨識指定的指標。如果需要，容器可以發出更多指標。 [KLL 草圖](https://datasketches.apache.org/docs/KLL/KLLSketch.html)是可辨識的草圖。自訂容器可以撰寫自己的表示法，但無法由 SageMaker AI 在視覺化中辨識。 依預設，分成 10 個儲存貯體將分佈具體化。這無法變更。 # 限制條件的結構描述 (constraints.json 檔案) constraints.json 檔案是用來表達資料集必須滿足的限制條件。Amazon SageMaker Model Monitor 容器可以使用 constraints.json 檔案來評估資料集。預先建置的容器能夠為基準資料集自動產生 constraints.json 檔案。如果您使用自有容器，則可以在容器中提供類似的功能，或者，您可以用其他方式建立 constraints.json 檔案。以下是預先建置的容器所使用的限制條件檔案的結構描述。使用自有容器可以採用相同的格式，或依需要來增強。 ``` { "version": 0, "features": [ { "name": "string", "inferred_type": "Integral" | "Fractional" | | "String" | "Unknown", "completeness": number, "num_constraints": { "is_non_negative": boolean }, "string_constraints": { "domains": [ "list of", "observed values", "for small cardinality" ] }, "monitoringConfigOverrides": {} } ], "monitoring_config": { "evaluate_constraints": "Enabled", "emit_metrics": "Enabled", "datatype_check_threshold": 0.1, "domain_content_threshold": 0.1, "distribution_constraints": { "perform_comparison": "Enabled", "comparison_threshold": 0.1, "comparison_method": "Simple"||"Robust", "categorical_comparison_threshold": 0.1, "categorical_drift_method": "LInfinity"||"ChiSquared" } } } ``` `monitoring_config` 物件包含用於監控功能工作的選項。下表描述各個選項。 監控限制條件 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/sagemaker/latest/dg/model-monitor-byoc-constraints.html) # 適用於使用自有容器的 CloudWatch 指標 在 `/opt/ml/processing/processingjobconfig.json` 檔案的 `Environment` 對應中，如果 `publish_cloudwatch_metrics` 值為 `Enabled`，容器程式碼會在此位置發出 Amazon CloudWatch 指標：`/opt/ml/output/metrics/cloudwatch`。 此檔案的結構描述緊密地以 `PutMetrics` API 為基礎。此處未指定命名空間。它預設為下列項目： + `For real-time endpoints: /aws/sagemaker/Endpoint/data-metrics` + `For batch transform jobs: /aws/sagemaker/ModelMonitoring/data-metrics` 但是，您可以指定維度。建議您至少增加以下維度： + 適用於即時端點的 `Endpoint` 和 `MonitoringSchedule` + 適用於批次轉換工作的 `MonitoringSchedule` 下列 JSON 程式碼片段顯示如何設定維度。 如果是即時端點，請參閱下列包含 `Endpoint` 和 `MonitoringSchedule` 維度的 JSON 程式碼片段： ``` { "MetricName": "", # Required "Timestamp": "2019-11-26T03:00:00Z", # Required "Dimensions" : [{"Name":"Endpoint","Value":"endpoint_0"},{"Name":"MonitoringSchedule","Value":"schedule_0"}] "Value": Float, # Either the Value or the StatisticValues field can be populated and not both. "StatisticValues": { "SampleCount": Float, "Sum": Float, "Minimum": Float, "Maximum": Float }, "Unit": "Count", # Optional } ``` 如果是批次轉換工作，請參閱下列包含 `MonitoringSchedule` 維度的 JSON 程式碼片段： ``` { "MetricName": "", # Required "Timestamp": "2019-11-26T03:00:00Z", # Required "Dimensions" : [{"Name":"MonitoringSchedule","Value":"schedule_0"}] "Value": Float, # Either the Value or the StatisticValues field can be populated and not both. "StatisticValues": { "SampleCount": Float, "Sum": Float, "Minimum": Float, "Maximum": Float }, "Unit": "Count", # Optional } ``` # 使用 CloudFormation 自訂資源建立即時端點的監控排程 如果您使用的是即時端點，您可以使用 CloudFormation 自訂資源來建立監控排程。自訂資源在 Python 中。若要部署，請參閱 [Python Lambda 部署](https://docs.aws.amazon.com/lambda/latest/dg/lambda-python-how-to-create-deployment-package.html)。 ## 自訂資源 首先，將自訂資源新增至您的 CloudFormation 範本。這會指向您在下一步中建立的 AWS Lambda 函式。 此資源可讓您自訂監控排程的參數。您可以在下列範例資源中修改 CloudFormation 資源和 Lambda 函數，以新增或移除更多參數。 ``` { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "MonitoringSchedule": { "Type": "Custom::MonitoringSchedule", "Version": "1.0", "Properties": { "ServiceToken": "arn:aws:lambda:us-west-2:111111111111:function:lambda-name", "ScheduleName": "YourScheduleName", "EndpointName": "YourEndpointName", "BaselineConstraintsUri": "s3://your-baseline-constraints/constraints.json", "BaselineStatisticsUri": "s3://your-baseline-stats/statistics.json", "PostAnalyticsProcessorSourceUri": "s3://your-post-processor/postprocessor.py", "RecordPreprocessorSourceUri": "s3://your-preprocessor/preprocessor.py", "InputLocalPath": "/opt/ml/processing/endpointdata", "OutputLocalPath": "/opt/ml/processing/localpath", "OutputS3URI": "s3://your-output-uri", "ImageURI": "111111111111.dkr.ecr.us-west-2.amazonaws.com/your-image", "ScheduleExpression": "cron(0 * ? * * *)", "PassRoleArn": "arn:aws:iam::111111111111:role/AmazonSageMaker-ExecutionRole" } } } } ``` ## Lambda 自訂資源程式碼 此 CloudFormation 自訂資源使用[自訂資源協助程式](https://github.com/aws-cloudformation/custom-resource-helper) AWS 程式庫，您可以使用 安裝 搭配 pip`pip install crhelper`。 在建立和刪除堆疊 CloudFormation 期間， 會叫用此 Lambda 函數。此 Lambda 函式負責建立及刪除監控排程，以及使用上一節所述之自訂資源中定義的參數。 ``` import boto3 import botocore import logging from crhelper import CfnResource from botocore.exceptions import ClientError logger = logging.getLogger(__name__) sm = boto3.client('sagemaker') # cfnhelper makes it easier to implement a CloudFormation custom resource helper = CfnResource() # CFN Handlers def handler(event, context): helper(event, context) @helper.create def create_handler(event, context): """ Called when CloudFormation custom resource sends the create event """ create_monitoring_schedule(event) @helper.delete def delete_handler(event, context): """ Called when CloudFormation custom resource sends the delete event """ schedule_name = get_schedule_name(event) delete_monitoring_schedule(schedule_name) @helper.poll_create def poll_create(event, context): """ Return true if the resource has been created and false otherwise so CloudFormation polls again. """ schedule_name = get_schedule_name(event) logger.info('Polling for creation of schedule: %s', schedule_name) return is_schedule_ready(schedule_name) @helper.update def noop(): """ Not currently implemented but crhelper will throw an error if it isn't added """ pass # Helper Functions def get_schedule_name(event): return event['ResourceProperties']['ScheduleName'] def create_monitoring_schedule(event): schedule_name = get_schedule_name(event) monitoring_schedule_config = create_monitoring_schedule_config(event) logger.info('Creating monitoring schedule with name: %s', schedule_name) sm.create_monitoring_schedule( MonitoringScheduleName=schedule_name, MonitoringScheduleConfig=monitoring_schedule_config) def is_schedule_ready(schedule_name): is_ready = False schedule = sm.describe_monitoring_schedule(MonitoringScheduleName=schedule_name) status = schedule['MonitoringScheduleStatus'] if status == 'Scheduled': logger.info('Monitoring schedule (%s) is ready', schedule_name) is_ready = True elif status == 'Pending': logger.info('Monitoring schedule (%s) still creating, waiting and polling again...', schedule_name) else: raise Exception('Monitoring schedule ({}) has unexpected status: {}'.format(schedule_name, status)) return is_ready def create_monitoring_schedule_config(event): props = event['ResourceProperties'] return { "ScheduleConfig": { "ScheduleExpression": props["ScheduleExpression"], }, "MonitoringJobDefinition": { "BaselineConfig": { "ConstraintsResource": { "S3Uri": props['BaselineConstraintsUri'], }, "StatisticsResource": { "S3Uri": props['BaselineStatisticsUri'], } }, "MonitoringInputs": [ { "EndpointInput": { "EndpointName": props["EndpointName"], "LocalPath": props["InputLocalPath"], } } ], "MonitoringOutputConfig": { "MonitoringOutputs": [ { "S3Output": { "S3Uri": props["OutputS3URI"], "LocalPath": props["OutputLocalPath"], } } ], }, "MonitoringResources": { "ClusterConfig": { "InstanceCount": 1, "InstanceType": "ml.t3.medium", "VolumeSizeInGB": 50, } }, "MonitoringAppSpecification": { "ImageUri": props["ImageURI"], "RecordPreprocessorSourceUri": props['PostAnalyticsProcessorSourceUri'], "PostAnalyticsProcessorSourceUri": props['PostAnalyticsProcessorSourceUri'], }, "StoppingCondition": { "MaxRuntimeInSeconds": 300 }, "RoleArn": props["PassRoleArn"], } } def delete_monitoring_schedule(schedule_name): logger.info('Deleting schedule: %s', schedule_name) try: sm.delete_monitoring_schedule(MonitoringScheduleName=schedule_name) except ClientError as e: if e.response['Error']['Code'] == 'ResourceNotFound': logger.info('Resource not found, nothing to delete') else: logger.error('Unexpected error while trying to delete monitoring schedule') raise e ```