AWS Glue-CloudWatch事件：如何在作业触发事件中获取自定义参数

默认情况下，AWS Glue发送到EventBridge的「Glue Job State Change」事件只会包含作业名称、运行ID、状态这类基础元数据，不会自动带上你通过aws glue start-job-run传入的自定义arguments参数。要解决这个问题，你可以参考以下两种实用方案：

方案一：在Glue作业中主动发送包含自定义参数的事件 #

这种方法让你完全控制事件的内容，直接把自定义参数嵌入到发送给EventBridge的事件里：

1. 给Glue作业执行角色添加权限：确保作业使用的IAM角色拥有events:PutEvents的权限，这样作业代码才能调用EventBridge的API发送事件。
2. 在作业代码中添加事件发送逻辑：当作业完成（成功或失败）时，获取传入的自定义参数，构造事件并发送到EventBridge。

举个Python代码示例（Glue作业中使用）：

import sys
import json
import os
import boto3
from awsglue.utils import getResolvedOptions

def send_custom_event_to_eventbridge(job_name, job_run_id, custom_args, state):
    eventbridge_client = boto3.client('events')
    event_detail = {
        "jobName": job_name,
        "jobRunId": job_run_id,
        "state": state,
        "customArguments": custom_args,
        "message": f"Job run {state} successfully"
    }

    try:
        response = eventbridge_client.put_events(
            Entries=[
                {
                    "Source": "custom.glue.job.status",
                    "DetailType": "Custom Glue Job State Change",
                    "Detail": json.dumps(event_detail),
                    "Region": os.environ["AWS_REGION"]
                }
            ]
        )
        print(f"Sent custom event: {response}")
    except Exception as e:
        print(f"Failed to send event: {str(e)}")

# 作业主逻辑
if __name__ == "__main__":
    # 获取传入的自定义参数
    args = getResolvedOptions(sys.argv, ['keyName1', 'keyName2'])
    custom_arguments = {k: args[k] for k in args}

    # 这里写你的作业业务逻辑
    # ...

    # 作业完成后发送事件（示例为成功状态，可根据实际情况处理失败状态）
    job_name = os.environ["JOB_NAME"]
    job_run_id = os.environ["JOB_RUN_ID"]
    send_custom_event_to_eventbridge(job_name, job_run_id, custom_arguments, "SUCCEEDED")

3. 创建新的EventBridge规则：监听你自定义的Source（custom.glue.job.status）和DetailType（Custom Glue Job State Change），这样就能收到包含自定义参数的事件了。

方案二：用Lambda函数补充获取自定义参数 #

如果你不想修改Glue作业代码，可以通过Lambda在事件触发时主动拉取作业运行的详细信息：

1. 创建Lambda函数：编写Lambda代码，当收到原始的Glue状态变更事件时，调用Glue的get_job_run API获取作业运行的完整数据（包含自定义参数）。

Lambda代码示例：

import boto3

glue_client = boto3.client('glue')

def lambda_handler(event, context):
    # 从原始事件中提取作业名称和运行ID
    job_name = event['detail']['jobName']
    job_run_id = event['detail']['jobRunId']

    try:
        # 获取作业运行的详细信息，包含自定义arguments
        job_run_details = glue_client.get_job_run(
            JobName=job_name,
            RunId=job_run_id
        )
        custom_arguments = job_run_details['JobRun']['Arguments']

        # 这里可以根据需求处理数据，比如转发到其他服务、存储到S3等
        print(f"Custom arguments for job run {job_run_id}: {custom_arguments}")

        return {
            'statusCode': 200,
            'body': {
                'original_event': event,
                'custom_arguments': custom_arguments
            }
        }
    except Exception as e:
        print(f"Error fetching job run details: {str(e)}")
        raise e

2. 配置Lambda权限：确保Lambda的执行角色拥有glue:GetJobRun的权限，允许它调用Glue的API。
3. 更新EventBridge规则：把这个Lambda函数设置为事件规则的目标，这样每次原始事件触发时，Lambda都会自动获取并处理包含自定义参数的作业数据。

注意事项 #

• 原始的「Glue Job State Change」事件结构是AWS固定的，无法直接修改让它包含自定义参数，所以必须通过上述两种方式实现需求。
• 方案一更适合需要直接获取包含参数的事件的场景，方案二则适合不想改动作业代码的场景。

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