如何捕获SageMaker作业错误并通过SES、SNS发送邮件通知

方案一：基于EventBridge事件触发（推荐，无额外资源消耗） #

• 第一步：创建EventBridge规则，事件源选择aws.sagemaker，事件类型匹配SageMaker处理任务状态变更，事件过滤规则设置为detail.CurrentProcessingJobStatus = Failed，仅任务失败时触发规则。
• 第二步：错误信息可直接从事件Payload的detail.FailureReason字段提取，该字段返回内容和SageMaker控制台显示的失败原因完全一致，涵盖算法错误、API参数错误、权限不足等所有异常场景的精准描述。
• 第三步：配置EventBridge规则的下游目标：
  • 轻量实现：目标选择SNS主题，提前给主题订阅接收通知的邮箱，触发后自动推送完整失败事件到所有订阅邮箱。
  • 自定义实现：目标选择独立Lambda函数，在函数中格式化错误信息、附加CloudWatch日志链接等内容后，调用SES或者SNS发送定制化邮件。

方案二：原有创建任务的Lambda内同步实现 #

• 调用create_processing_job接口创建任务后，通过describe_processing_job接口轮询任务状态，轮询间隔建议设置为30~60秒，避免触发SageMaker API限流。
• 当接口返回的ProcessingJobStatus字段值为Failed时，直接提取返回体中的FailureReason字段内容，即为精准错误信息。
• 提取错误信息后，直接在当前Lambda中调用SNS/SES发送通知即可，需提前为Lambda配置对应服务的发送权限。

参考代码片段 #

import boto3
import time

sagemaker_client = boto3.client("sagemaker")
sns_client = boto3.client("sns")
# 替换为实际SNS主题ARN
SNS_NOTICE_TOPIC_ARN = "arn:aws-cn:sns:cn-north-1:123456789012:sagemaker_fail_notice"

def check_processing_job_status(job_name: str):
    while True:
        job_detail = sagemaker_client.describe_processing_job(ProcessingJobName=job_name)
        job_status = job_detail["ProcessingJobStatus"]
        
        if job_status == "Completed":
            return True
        elif job_status == "Failed":
            fail_reason = job_detail["FailureReason"]
            # 发送失败通知
            sns_client.publish(
                TopicArn=SNS_NOTICE_TOPIC_ARN,
                Subject=f"警报：SageMaker处理任务 {job_name} 运行失败",
                Message=f"任务失败详情如下：\n{fail_reason}"
            )
            return False
        # 非终态等待后重试
        time.sleep(60)

注意事项 #

• 两种方案均需提前配置对应IAM权限：EventBridge需要SNS/Lambda的调用权限，Lambda需要SageMaker的任务查询权限、SNS/SES的发送权限。
• 若需要更详细的运行日志，可通过任务ID拉取对应CloudWatch Logs的内容，附加到邮件通知中。

内容的提问来源于stack exchange，提问作者rishab ajain445