基于AWS Glue Workflow实现ETL失败时Amazon SES邮件告警

1. 配置Workflow错误分支 #

给每个ETL任务设置失败触发规则，直接指向一个专门的告警任务（建议用Python Shell类型，轻量易实现）。在Glue Workflow可视化编辑器里选中ETL任务，设置“失败时”的下一个任务为告警任务即可。

2. 获取失败任务的名称和错误信息 #

Glue Workflow会记录整个运行的上下文，告警任务里可以通过Glue API拉取当前Workflow的运行详情：

• 利用Glue任务环境变量GLUE_WORKFLOW_NAME和GLUE_WORKFLOW_RUN_ID，调用get_workflow_run接口拿到运行数据
• 从返回结果的FailedJobs字段里提取失败任务名、错误详情等信息

3. 用SES发送告警邮件 #

在告警任务的Python脚本里调用SES API发邮件，步骤如下：

• 确保Glue执行角色拥有ses:SendEmail权限，且SES已验证发件邮箱和收件方的域名/邮箱
• 示例代码片段：

import boto3
import os
from botocore.exceptions import ClientError

def send_failure_alert(subject, content):
    ses = boto3.client('ses', region_name='your-region')
    sender = 'alerts@your-domain.com'
    recipients = ['business-staff@your-domain.com']
    try:
        ses.send_email(
            Source=sender,
            Destination={'ToAddresses': recipients},
            Message={
                'Subject': {'Data': subject},
                'Body': {'Text': {'Data': content}}
            }
        )
    except ClientError as e:
        print(f"邮件发送失败: {e.response['Error']['Message']}")
    else:
        print("告警邮件已发送")

# 拉取Workflow运行信息
glue = boto3.client('glue', region_name='your-region')
run_details = glue.get_workflow_run(
    WorkflowName=os.environ['GLUE_WORKFLOW_NAME'],
    RunId=os.environ['GLUE_WORKFLOW_RUN_ID']
)

# 解析失败任务
failed_jobs = run_details['Run']['FailedJobs']
if failed_jobs:
    alert_subject = f"Glue Workflow {os.environ['GLUE_WORKFLOW_NAME']} 任务失败通知"
    alert_content = "以下ETL任务执行失败:\n"
    for job in failed_jobs:
        alert_content += f"\n任务名称: {job['JobName']}"
        alert_content += f"\n错误信息: {job['ErrorDetails']['ErrorMessage']}"
        alert_content += "\n---"
    send_failure_alert(alert_subject, alert_content)

调用其他Glue任务 #

完全可以，直接在当前Glue任务（不管是Spark还是Python Shell）里用boto3调用start_job_run API启动其他Glue任务，不需要Glue端点或额外安全组规则。只要当前Glue角色有glue:StartJobRun权限就行。示例代码：

import boto3
glue = boto3.client('glue')
# 启动另一个Glue任务
run_response = glue.start_job_run(
    JobName='target-glue-job',
    Arguments={'--custom-param': 'your-value'}
)
print(f"已启动任务，Run ID: {run_response['JobRunId']}")

调用外部Python脚本 #

也可以实现，不需要额外端点或安全组，只要Glue角色有对应S3读取权限：

• 把脚本存到S3，在当前任务里下载到临时目录，再通过importlib加载执行
• 示例代码：

import boto3
import importlib.util
import os

s3 = boto3.client('s3')
bucket = 'your-bucket'
script_path = 'scripts/your-script.py'
local_tmp_path = f'/tmp/{os.path.basename(script_path)}'

# 从S3下载脚本
s3.download_file(bucket, script_path, local_tmp_path)

# 加载并执行脚本
spec = importlib.util.spec_from_file_location('external_script', local_tmp_path)
external_script = importlib.util.module_from_spec(spec)
spec.loader.exec_module(external_script)

# 调用脚本中的函数
external_script.your_function()

如果脚本依赖第三方库，可以通过Glue任务的--additional-python-modules参数指定安装。

内容的提问来源于stack exchange，提问作者B Kiran