A collection of notes and resources
モデル推論をモニタリングする
from sagemaker.model_monitor import DataCaptureConfig, DefaultModelMonitor
from sagemaker.model_monitor.dataset_format import DatasetFormat
# データキャプチャ設定
data_capture_config = DataCaptureConfig(
enable_capture=True,
sampling_percentage=100,
destination_s3_uri=f's3://{bucket}/data-capture'
)
# エンドポイントにデータキャプチャを有効化
predictor = model.deploy(
initial_instance_count=1,
instance_type='ml.m5.xlarge',
data_capture_config=data_capture_config
)
# ベースライン作成
baseline_processor = DefaultModelMonitor(
role=role,
instance_count=1,
instance_type='ml.m5.xlarge',
volume_size_in_gb=20,
max_runtime_in_seconds=3600
)
baseline_processor.suggest_baseline(
baseline_dataset=f's3://{bucket}/training-data/baseline.csv',
dataset_format=DatasetFormat.csv(header=True),
output_s3_uri=f's3://{bucket}/baseline-results'
)
# モニタリングスケジュールの作成
monitor = DefaultModelMonitor(
role=role,
instance_count=1,
instance_type='ml.m5.xlarge',
volume_size_in_gb=20,
max_runtime_in_seconds=3600
)
monitor.create_monitoring_schedule(
monitor_schedule_name='data-quality-monitor',
endpoint_input=predictor.endpoint_name,
output_s3_uri=f's3://{bucket}/monitoring-results',
statistics=baseline_processor.baseline_statistics(),
constraints=baseline_processor.suggested_constraints(),
schedule_cron_expression='cron(0 * * * ? *)' # 毎時実行
)
from sagemaker.model_monitor import ModelQualityMonitor
model_quality_monitor = ModelQualityMonitor(
role=role,
instance_count=1,
instance_type='ml.m5.xlarge',
volume_size_in_gb=20,
max_runtime_in_seconds=3600
)
model_quality_monitor.create_monitoring_schedule(
monitor_schedule_name='model-quality-monitor',
endpoint_input=predictor.endpoint_name,
ground_truth_input=f's3://{bucket}/ground-truth/',
problem_type='BinaryClassification',
output_s3_uri=f's3://{bucket}/model-quality-results',
schedule_cron_expression='cron(0 0 * * ? *)' # 毎日実行
)
from sagemaker.model_monitor import BiasAnalysisConfig
bias_config = BiasAnalysisConfig(
bias_config_file_s3_uri=f's3://{bucket}/bias-config.json',
data_config_file_s3_uri=f's3://{bucket}/data-config.json'
)
bias_monitor.create_monitoring_schedule(
monitor_schedule_name='bias-drift-monitor',
endpoint_input=predictor.endpoint_name,
ground_truth_input=f's3://{bucket}/ground-truth/',
analysis_config=bias_config,
output_s3_uri=f's3://{bucket}/bias-results',
schedule_cron_expression='cron(0 0 * * ? *)'
)
from sagemaker.model_monitor import ModelExplainabilityMonitor
explainability_monitor = ModelExplainabilityMonitor(
role=role,
instance_count=1,
instance_type='ml.m5.xlarge'
)
explainability_monitor.create_monitoring_schedule(
monitor_schedule_name='explainability-monitor',
endpoint_input=predictor.endpoint_name,
output_s3_uri=f's3://{bucket}/explainability-results',
model_config_file_s3_uri=f's3://{bucket}/model-config.json',
schedule_cron_expression='cron(0 0 * * ? *)'
)
import boto3
logs = boto3.client('logs')
# ログストリームの作成
logs.create_log_stream(
logGroupName='/aws/sagemaker/Endpoints/production',
logStreamName='inference-errors'
)
# メトリクスフィルターの作成
logs.put_metric_filter(
logGroupName='/aws/sagemaker/Endpoints/production',
filterName='InferenceErrors',
filterPattern='[time, request_id, level=ERROR*, ...]',
metricTransformations=[
{
'metricName': 'InferenceErrors',
'metricNamespace': 'CustomMetrics',
'metricValue': '1',
'defaultValue': 0
}
]
)
cloudwatch = boto3.client('cloudwatch')
# エラー率のアラーム
cloudwatch.put_metric_alarm(
AlarmName='HighInferenceErrorRate',
ComparisonOperator='GreaterThanThreshold',
EvaluationPeriods=2,
MetricName='ModelInvocation4XXErrors',
Namespace='AWS/SageMaker',
Period=300,
Statistic='Sum',
Threshold=10.0,
ActionsEnabled=True,
AlarmActions=[sns_topic_arn],
AlarmDescription='Alert when inference error rate is high',
Dimensions=[
{
'Name': 'EndpointName',
'Value': 'production-endpoint'
},
{
'Name': 'VariantName',
'Value': 'AllTraffic'
}
]
)
# レイテンシーのアラーム
cloudwatch.put_metric_alarm(
AlarmName='HighModelLatency',
ComparisonOperator='GreaterThanThreshold',
EvaluationPeriods=3,
MetricName='ModelLatency',
Namespace='AWS/SageMaker',
Period=60,
Statistic='Average',
Threshold=1000.0, # 1秒
ActionsEnabled=True,
AlarmActions=[sns_topic_arn],
Dimensions=[
{
'Name': 'EndpointName',
'Value': 'production-endpoint'
},
{
'Name': 'VariantName',
'Value': 'AllTraffic'
}
]
)
from sagemaker import clarify
clarify_processor = clarify.SageMakerClarifyProcessor(
role=role,
instance_count=1,
instance_type='ml.m5.xlarge',
sagemaker_session=session
)
# データ設定
data_config = clarify.DataConfig(
s3_data_input_path=f's3://{bucket}/production-data/',
s3_output_path=f's3://{bucket}/clarify-output/',
label='target',
headers=['feature1', 'feature2', 'target'],
dataset_type='text/csv'
)
# バイアス設定
bias_config = clarify.BiasConfig(
label_values_or_threshold=[1],
facet_name='sensitive_attribute',
facet_values_or_threshold=[0]
)
# ドリフト分析実行
clarify_processor.run_post_training_bias(
data_config=data_config,
data_bias_config=bias_config,
model_config=model_config,
model_predicted_label_config=predicted_label_config
)
from sagemaker.session import production_variant
# バリアントA（現行モデル）
variant_a = production_variant(
model_name='model-v1',
instance_type='ml.m5.xlarge',
initial_instance_count=2,
variant_name='variant-a',
initial_weight=50
)
# バリアントB（新モデル）
variant_b = production_variant(
model_name='model-v2',
instance_type='ml.m5.xlarge',
initial_instance_count=2,
variant_name='variant-b',
initial_weight=50
)
# エンドポイント作成
predictor = sagemaker_session.endpoint_from_production_variants(
name='ab-test-endpoint',
production_variants=[variant_a, variant_b]
)
import boto3
from datetime import datetime, timedelta
cloudwatch = boto3.client('cloudwatch')
def get_variant_metrics(endpoint_name, variant_name, metric_name):
response = cloudwatch.get_metric_statistics(
Namespace='AWS/SageMaker',
MetricName=metric_name,
Dimensions=[
{'Name': 'EndpointName', 'Value': endpoint_name},
{'Name': 'VariantName', 'Value': variant_name}
],
StartTime=datetime.now() - timedelta(hours=1),
EndTime=datetime.now(),
Period=300,
Statistics=['Average', 'Sum']
)
return response
# バリアントA のメトリクス
metrics_a = get_variant_metrics('ab-test-endpoint', 'variant-a', 'ModelLatency')
# バリアントB のメトリクス
metrics_b = get_variant_metrics('ab-test-endpoint', 'variant-b', 'ModelLatency')
# 比較分析
print(f"Variant A average latency: {metrics_a}")
print(f"Variant B average latency: {metrics_b}")
import boto3
sagemaker = boto3.client('sagemaker')
# バリアントBが優れている場合、トラフィックを増やす
sagemaker.update_endpoint_weights_and_capacities(
EndpointName='ab-test-endpoint',
DesiredWeightsAndCapacities=[
{
'VariantName': 'variant-a',
'DesiredWeight': 20 # 20%
},
{
'VariantName': 'variant-b',
'DesiredWeight': 80 # 80%
}
]
)
import json
cloudwatch = boto3.client('cloudwatch')
dashboard_body = {
"widgets": [
{
"type": "metric",
"properties": {
"metrics": [
["AWS/SageMaker", "ModelLatency", {"stat": "Average"}],
[".", "Invocations", {"stat": "Sum"}],
[".", "ModelInvocation4XXErrors", {"stat": "Sum"}]
],
"period": 300,
"stat": "Average",
"region": "us-east-1",
"title": "Model Performance"
}
},
{
"type": "log",
"properties": {
"query": "SOURCE '/aws/sagemaker/Endpoints/production' | fields @timestamp, @message | filter @message like /ERROR/ | sort @timestamp desc | limit 20",
"region": "us-east-1",
"title": "Recent Errors"
}
}
]
}
cloudwatch.put_dashboard(
DashboardName='ML-Monitoring-Dashboard',
DashboardBody=json.dumps(dashboard_body)
)
| メトリクス | 説明 | アラーム閾値例 | |———–|——|—————| | ModelLatency | 推論レイテンシー | > 1000ms | | Invocations | 呼び出し回数 | < 100/分 (異常低下) | | ModelInvocation4XXErrors | クライアントエラー | > 5% | | ModelInvocation5XXErrors | サーバーエラー | > 1% | | CPUUtilization | CPU使用率 | > 80% | | MemoryUtilization | メモリ使用率 | > 85% | | DiskUtilization | ディスク使用率 | > 90% |