如何在Glue作业中获取当前运行结束时间并上报云指标？

一、获取本次运行实例的结束时间 #

get_job_run API仅在作业完成后才会返回结束时间，运行过程中无法通过该接口获取，因此需要在作业代码内自行处理：

• 代码内主动记录时间戳
  在作业启动时记录开始时间，然后在正常结束逻辑或异常捕获块中记录结束时间，同时可以通过Glue内置参数获取当前运行ID：

  import datetime
  import sys
  from awsglue.utils import getResolvedOptions
  
  # 获取当前作业运行ID
  args = getResolvedOptions(sys.argv, ['JOB_RUN_ID'])
  job_run_id = args['JOB_RUN_ID']
  
  # 记录启动时间
  start_time = datetime.datetime.now()
  
  try:
      # 作业核心业务逻辑
      print("执行作业任务...")
      # ...
  except Exception as e:
      # 异常时记录结束时间
      end_time = datetime.datetime.now()
      print(f"作业异常终止，运行ID: {job_run_id}，结束时间: {end_time}")
      # 此处可插入指标上报逻辑
      raise
  else:
      # 正常结束时记录结束时间
      end_time = datetime.datetime.now()
      print(f"作业正常完成，运行ID: {job_run_id}，结束时间: {end_time}")
      # 此处可插入指标上报逻辑
  

• 作业完成后通过外部触发获取
  如果不需要在作业运行过程中拿到结束时间，可配置CloudWatch Events或Glue作业完成触发器，触发Lambda函数调用get_job_run接口，此时就能获取包含结束时间的完整运行信息。

二、将结束时间和错误信息输出到云指标（以CloudWatch为例） #

使用boto3调用CloudWatch的put_metric_data接口，上报自定义指标：

1. 上报结束时间 #

将结束时间转为Unix时间戳作为指标值，配合作业名、运行ID作为维度区分：

import boto3

cloudwatch = boto3.client('cloudwatch')
# 转换结束时间为Unix时间戳（秒级）
end_timestamp = int(end_time.timestamp())

cloudwatch.put_metric_data(
    Namespace='GlueJobCustomMetrics',
    MetricData=[
        {
            'MetricName': 'JobEndTimestamp',
            'Dimensions': [
                {'Name': 'JobName', 'Value': '你的Glue作业名称'},
                {'Name': 'JobRunId', 'Value': job_run_id}
            ],
            'Value': end_timestamp,
            'Unit': 'None'
        }
    ]
)

2. 上报错误信息 #

错误信息为字符串，无法直接作为指标值，可通过两种方式处理：

• 上报错误类型/自定义错误码
  捕获异常时提取错误类型，映射为自定义数值码上报，同时将详细错误信息输出到CloudWatch Logs（Glue作业默认日志输出到CloudWatch）：

  except Exception as e:
      end_time = datetime.datetime.now()
      error_type = type(e).__name__
      # 自定义错误码映射，比如KeyError=1，ValueError=2等
      error_code = 1 if error_type == 'KeyError' else 2
  
      cloudwatch.put_metric_data(
          Namespace='GlueJobCustomMetrics',
          MetricData=[
              {
                  'MetricName': 'JobErrorCode',
                  'Dimensions': [
                      {'Name': 'JobName', 'Value': '你的Glue作业名称'},
                      {'Name': 'JobRunId', 'Value': job_run_id},
                      {'Name': 'ErrorType', 'Value': error_type}
                  ],
                  'Value': error_code,
                  'Unit': 'None'
              }
          ]
      )
      # 打印错误详情到日志
      print(f"错误详情: {str(e)}")
      raise
  

• 上报作业状态指标
  用0标记成功、1标记失败，将状态作为指标值上报，错误信息关联到对应运行ID的日志中，通过CloudWatch指标与日志联动查看：

  # 正常结束时上报状态0
  cloudwatch.put_metric_data(
      Namespace='GlueJobCustomMetrics',
      MetricData=[
          {
              'MetricName': 'JobStatus',
              'Dimensions': [
                  {'Name': 'JobName', 'Value': '你的Glue作业名称'},
                  {'Name': 'JobRunId', 'Value': job_run_id}
              ],
              'Value': 0,
              'Unit': 'None'
          }
      ]
  )
  
  # 异常结束时上报状态1
  cloudwatch.put_metric_data(
      Namespace='GlueJobCustomMetrics',
      MetricData=[
          {
              'MetricName': 'JobStatus',
              'Dimensions': [
                  {'Name': 'JobName', 'Value': '你的Glue作业名称'},
                  {'Name': 'JobRunId', 'Value': job_run_id}
              ],
              'Value': 1,
              'Unit': 'None'
          }
      ]
  )
  

注意事项 #

• 确保Glue作业的IAM角色拥有cloudwatch:PutMetricData权限，否则无法完成指标上报。
• 作业运行ID通过Glue内置参数JOB_RUN_ID自动获取，无需手动传入。

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