如何在Lambda触发Glue作业后判断其是否执行成功

要实现Lambda等待Glue作业完成后再返回结果，核心逻辑是启动作业后持续轮询作业状态，直到作业进入终态（成功/失败/停止），再根据终态结果返回对应响应给Step Functions。

修改后的Lambda代码 #

import json
import boto3
import time

def lambda_handler(event, context):
    glue_client = boto3.client('glue')
    
    # 启动Glue作业
    try:
        start_response = glue_client.start_job_run(
            JobName='job1',
            Arguments={'--parametro1': 'par 1'}
        )
        job_run_id = start_response['JobRunId']
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps(f"启动Glue作业失败: {str(e)}")
        }
    
    # 轮询作业状态，直到进入终态
    while True:
        try:
            run_response = glue_client.get_job_run(
                JobName='job1',
                RunId=job_run_id
            )
            job_status = run_response['JobRun']['JobRunState']
            
            # 检查是否为终态
            if job_status in ['SUCCEEDED', 'FAILED', 'STOPPED']:
                break
            
            # 非终态则等待一段时间后再次查询
            time.sleep(30)
        except Exception as e:
            return {
                'statusCode': 500,
                'body': json.dumps(f"查询Glue作业状态失败: {str(e)}")
            }
    
    # 根据作业终态返回对应结果
    if job_status == 'SUCCEEDED':
        return {
            'statusCode': 200,
            'body': json.dumps(f"Glue作业执行成功，RunId: {job_run_id}")
        }
    else:
        error_message = run_response['JobRun'].get('ErrorMessage', '无错误信息')
        return {
            'statusCode': 500,
            'body': json.dumps(f"Glue作业执行失败，状态: {job_status}，错误信息: {error_message}，RunId: {job_run_id}")
        }

关键注意事项 #

• Lambda超时设置：必须确保Lambda的超时时间大于Glue作业的最长预期运行时间，否则Lambda会提前终止，无法获取最终结果。比如Glue作业最长跑30分钟，Lambda超时可设为35分钟。
• 轮询间隔：示例中用30秒间隔，可根据实际需求调整，避免过于频繁调用API触发限流。
• 异常处理：代码捕获了启动作业和查询状态时的异常，确保任何环节出错都能返回明确的错误信息给Step Functions。
• Step Functions集成：Step Functions可根据Lambda返回的statusCode做分支处理，比如200时执行下一步，500时触发重试或错误告警流程。

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