翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。 # 高度なトピック 以下のセクションでは、前処理スクリプトと後処理スクリプトを使用してモニタリングをカスタマイズする方法、独自のコンテナを構築する方法、および 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 コンテナは、表形式または `{“prediction0”: 1.0, “prediction1” : 2.1}` のようにフラット化された JSON 構造でのみ動作します。次のような前処理スクリプトを使用して、配列を正しい 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()) ``` ## カスタムサンプリング 前処理スクリプトにカスタムサンプリング戦略を適用することもできます。そのためには、Model Monitor のファーストパーティのビルド済みコンテナを設定して、指定したサンプリングレートに従って一部のレコードを無視するようにします。次の例では、ハンドラーはハンドラー呼び出しの 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 に記録された例外メッセージを確認してデバッグします。`preprocess_handler` インターフェイスを通じて CloudWatch のロガーにアクセスできます。スクリプトから必要な情報をすべて 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 には、エンドポイントまたはバッチ変換ジョブからキャプチャした表形式のデータセットのデータを分析する機能を備えた、ビルド済みコンテナが用意されています。独自のコンテナを持ち込む場合は、Model Monitor が提供する拡張ポイントを活用できます。 内部的には、`MonitoringSchedule` を作成すると、Model Monitor が処理ジョブを最終的に開始します。したがってコンテナは、「[独自の処理コンテナを構築する方法 (高度なシナリオ)](build-your-own-processing-container.md)」トピックに記載されている処理ジョブのコントラクトを認識する必要があります。Model Monitor はスケジュールに従って自動的に処理ジョブを開始することに注意してください。呼び出し中に、Model Monitor は追加の環境変数を設定し、スケジュールされた特定のモニタリング実行のデータ処理に十分なコンテキストがコンテナに含まれるようにします。コンテナ入力の詳細については、「[コンテナコントラクトの入力](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" } ``` パラメータ | Parameter Name | 説明 | | --- | --- | | dataset\$1format | `Endpoint` によって支援された `MonitoringSchedule` から開始されたジョブでは、これは `endpointInput`、`endpointOutput`、またはその両方のキャプチャインデックスを持つ `sageMakerCaptureJson` になります。バッチ変換ジョブの場合、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 | この分析の実行期間。たとえば、2020 年 2 月 20 日の 05:00 UTC に実行するようにスケジュールされたジョブの場合、`start_time`: は 2020-02-19T06:00:00Z、`end_time`: は 202020-02-20T 05:00:00 Z になります。 | | baseline\$1constraints: | ` BaselineConfig.ConstraintResource.S3Uri` で指定されているベースライン制約ファイルのローカルパス。これは、このパラメータが `CreateMonitoringSchedule` リクエストで指定された場合にのみ使用できます。 | | baseline\$1statistics | `BaselineConfig.StatisticsResource.S3Uri` で指定したベースライン統計ファイルへのローカルパス。これは、このパラメータが `CreateMonitoringSchedule` リクエストで指定された場合にのみ使用できます。 | # コンテナコントラクトの出力 コンテナは、`*dataset_source*` パスで使用可能なデータを分析し、`*output_path*.` 内のパスにレポートを書き込むことができます。コンテナコードは、ニーズに合った任意のレポートを書くことができます。 次の構造と契約を使用すると、特定の出力ファイルは可視化と API で SageMaker AI によって特別に処理されます。これは、表形式のデータセットにのみ適用されます。 表形式データセットの出力ファイル | ファイル名 | 説明 | | --- | --- | | statistics.json | このファイルには、分析対象のデータセット内の各フィーチャに対する列状の統計が含まれます。このファイルのスキーマは次のセクションで確認できます。 | | constraints.json | このファイルには、確認されたフィーチャの制約が含まれます。このファイルのスキーマは次のセクションで確認できます。 | | constraints\$1violations.json | このファイルには、`baseline_constaints` および `baseline_statistics` パスで指定されたベースライン統計および制約ファイルと比較して、この現在のデータセットで検出された違反のリストが含まれています。 | さらに、`publish_cloudwatch_metrics` 値が `"Enabled"` である場合、コンテナコードは `/opt/ml/output/metrics/cloudwatch` に Amazon 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) が使用するバケットを設定します。KLL は、遅延圧縮スキームを持つ非常にコンパクトな分位数スケッチです。 ``` { "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/ja_jp/sagemaker/latest/dg/model-monitor-byoc-constraints.html) # 独自のコンテナを取り込む場合の CloudWatch メトリクス `/opt/ml/processing/processingjobconfig.json` ファイル内の `Environment` マップで `publish_cloudwatch_metrics` 値が `Enabled` である場合、コンテナコードは `/opt/ml/output/metrics/cloudwatch` に Amazon CloudWatch メトリクスを出力します。 このファイルのスキーマは、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 カスタムリソースは、 を使用して pip でインストールできる[カスタムリソースヘルパー](https://github.com/aws-cloudformation/custom-resource-helper) AWS ライブラリを使用します`pip install crhelper`。 この Lambda 関数は、スタックの作成と削除 CloudFormation 中に によって呼び出されます。この 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 ```