如何用Python BOTO3程序化验证AWS Lambda与Glue Job的运行状态

核心思路 #

分别通过Boto3调用AWS Lambda和Glue的官方API，结合CloudWatch Logs提取最新运行记录的状态，全程程序化实现，无需依赖CloudWatch告警机制。

具体实现步骤 #

1. 验证Lambda函数状态 #

Lambda的执行状态需从对应CloudWatch日志组中提取：

• 初始化boto3.client('logs')，目标日志组格式固定为/aws/lambda/<函数名>
• 按事件时间倒序获取最新日志流，再提取该流内的最新日志事件
• 通过包含REPORT的日志行判断执行结果，或直接匹配Error、Task timed out等关键词识别失败场景

2. 验证Glue任务状态 #

Glue提供直接查询任务运行记录的API，无需额外日志解析：

• 初始化boto3.client('glue')，调用get_job_runs(JobName=<任务名>, MaxResults=1)获取最新一次运行记录
• 读取返回结果中JobRuns[0]['JobRunState']字段，SUCCEEDED即为成功，FAILED/STOPPED等标记为失败

完整Python脚本示例 #

import boto3
from datetime import datetime

def check_lambda_last_status(lambda_function_name):
    logs_client = boto3.client('logs')
    log_group_name = f'/aws/lambda/{lambda_function_name}'
    
    try:
        # 获取最新日志流
        log_streams = logs_client.describe_log_streams(
            logGroupName=log_group_name,
            orderBy='LastEventTime',
            descending=True,
            limit=1
        )['logStreams']
        if not log_streams:
            return f"{lambda_function_name}: 无运行记录"
        
        latest_stream = log_streams[0]
        # 获取该流最新10条日志（确保覆盖REPORT行）
        events = logs_client.get_log_events(
            logGroupName=log_group_name,
            logStreamName=latest_stream['logStreamName'],
            startFromHead=False,
            limit=10
        )['events']
        
        # 倒序遍历找关键状态
        for event in reversed(events):
            msg = event['message']
            if 'REPORT' in msg:
                if 'Error' in msg or 'Task timed out' in msg:
                    return f"{lambda_function_name}: 失败"
                else:
                    return f"{lambda_function_name}: 成功"
            elif 'Error' in msg:
                return f"{lambda_function_name}: 失败"
        return f"{lambda_function_name}: 无法解析状态"
    except Exception as e:
        return f"{lambda_function_name}: 查询失败 - {str(e)}"

def check_glue_last_status(glue_job_name):
    glue_client = boto3.client('glue')
    try:
        job_runs = glue_client.get_job_runs(
            JobName=glue_job_name,
            MaxResults=1
        )['JobRuns']
        if not job_runs:
            return f"{glue_job_name}: 无运行记录"
        
        latest_run = job_runs[0]
        run_state = latest_run['JobRunState']
        run_date = datetime.fromtimestamp(latest_run['StartedOn'].timestamp()).date()
        # 可选：验证是否为今日运行任务
        if run_date != datetime.today().date():
            return f"{glue_job_name}: 今日未运行（上次状态：{run_state}）"
        
        return f"{glue_job_name}: {run_state}"
    except Exception as e:
        return f"{glue_job_name}: 查询失败 - {str(e)}"

if __name__ == "__main__":
    # 替换为你的Lambda函数列表
    lambda_functions = ['daily-data-sync', 'user-event-process', 'report-generator']
    # 替换为你的Glue任务列表
    glue_jobs = ['raw-data-ingest', 'dwh-transform-job']
    
    print("=== Lambda函数运行状态 ===")
    for func in lambda_functions:
        print(check_lambda_last_status(func))
    
    print("\n=== Glue任务运行状态 ===")
    for job in glue_jobs:
        print(check_glue_last_status(job))

注意事项 #

• 权限配置：运行脚本的IAM角色需具备以下权限：
  • logs:DescribeLogStreams、logs:GetLogEvents
  • glue:GetJobRuns
  • 若需动态获取函数/任务列表，需额外添加lambda:ListFunctions、glue:GetJobs
• 时区适配：脚本默认使用UTC时间，若需匹配本地时区，可调整datetime转换逻辑
• 异常扩展：可根据需求补充日志组不存在、任务未创建等场景的异常处理
• 定时执行：可将脚本部署为定时Lambda函数，每日固定时间执行后将结果写入S3或发送通知

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