如何用AWS SDK、Lambda或Python脚本获取指定Job Queue的失败AWS Batch任务？

完全可以通过Python脚本（结合AWS SDK for Python boto3）实现需求，以下是具体步骤和代码示例：

一、获取指定Job Queue的失败任务名称 #

使用boto3的batch客户端调用list_jobs接口，分页获取目标队列中状态为FAILED的任务，提取任务名称。添加时间范围过滤，适配每日运行场景，避免重复获取全量历史数据。

import boto3
from datetime import datetime, timedelta

def get_failed_batch_jobs(queue_name):
    batch_client = boto3.client('batch')
    failed_jobs = []
    next_token = None

    # 过滤过去24小时内的失败任务
    start_time = (datetime.now() - timedelta(hours=24)).timestamp() * 1000

    while True:
        kwargs = {
            'jobQueue': queue_name,
            'jobStatus': 'FAILED',
            'createdAfter': int(start_time)
        }
        if next_token:
            kwargs['nextToken'] = next_token
        
        response = batch_client.list_jobs(**kwargs)
        failed_jobs.extend([job['jobName'] for job in response['jobSummaryList']])
        
        next_token = response.get('nextToken')
        if not next_token:
            break
    
    return failed_jobs

# 调用示例
queue_name = 'your-target-job-queue'
failed_job_names = get_failed_batch_jobs(queue_name)

二、将结果上传至S3或RDS #

1. 上传至S3 #

将失败任务名称保存为CSV格式，通过boto3的s3客户端上传到指定桶：

import csv
from io import StringIO

def upload_to_s3(job_names, bucket_name, s3_key):
    s3_client = boto3.client('s3')
    csv_buffer = StringIO()
    writer = csv.writer(csv_buffer)
    writer.writerow(['failed_job_name'])
    for name in job_names:
        writer.writerow([name])
    
    s3_client.put_object(
        Bucket=bucket_name,
        Key=s3_key,
        Body=csv_buffer.getvalue(),
        ContentType='text/csv'
    )

# 调用示例
bucket = 'your-s3-bucket'
key = f'failed-batch-jobs/{datetime.now().strftime("%Y-%m-%d")}.csv'
upload_to_s3(failed_job_names, bucket, key)

2. 写入至RDS（以PostgreSQL为例） #

使用psycopg2连接RDS实例，将任务名称插入指定表（需提前创建表结构）：

import psycopg2

def write_to_rds(job_names, db_config):
    conn = psycopg2.connect(
        host=db_config['host'],
        database=db_config['dbname'],
        user=db_config['user'],
        password=db_config['password']
    )
    cursor = conn.cursor()
    
    # 批量插入数据
    insert_query = "INSERT INTO failed_batch_jobs (job_name, created_at) VALUES (%s, %s)"
    current_time = datetime.now()
    data = [(name, current_time) for name in job_names]
    
    cursor.executemany(insert_query, data)
    conn.commit()
    
    cursor.close()
    conn.close()

# 调用示例
db_config = {
    'host': 'your-rds-endpoint',
    'dbname': 'your-db-name',
    'user': 'your-db-user',
    'password': 'your-db-password'
}
write_to_rds(failed_job_names, db_config)

三、配置每日自动运行 #

1. 部署为AWS Lambda #

将上述代码整合为Lambda handler函数，注意：

• 确保Lambda执行角色拥有以下权限：
  • batch:ListJobs（针对目标Job Queue）
  • s3:PutObject（针对目标S3桶）或RDS连接权限（如rds-db:connect）
  • logs:CreateLogGroup、logs:CreateLogStream、logs:PutLogEvents（用于日志输出）
• 若使用RDS，需确保Lambda在VPC内可访问RDS实例，或RDS配置为公网可访问（生产环境不推荐后者）

Lambda handler示例：

import boto3
from datetime import datetime, timedelta
import csv
from io import StringIO

def lambda_handler(event, context):
    # 1. 获取失败任务
    queue_name = 'your-target-job-queue'
    failed_job_names = get_failed_batch_jobs(queue_name)
    
    if not failed_job_names:
        return {'statusCode': 200, 'body': 'No failed jobs found'}
    
    # 2. 上传至S3（替换为写入RDS逻辑即可）
    bucket = 'your-s3-bucket'
    key = f'failed-batch-jobs/{datetime.now().strftime("%Y-%m-%d")}.csv'
    upload_to_s3(failed_job_names, bucket, key)
    
    return {'statusCode': 200, 'body': f'Uploaded {len(failed_job_names)} failed jobs to S3'}

# 包含前面的get_failed_batch_jobs和upload_to_s3函数

2. 设置每日定时触发 #

通过Amazon EventBridge设置定时规则：

• 进入EventBridge控制台，创建新规则，选择计划类型
• 设置 cron 表达式为 0 0 * * ? *（每日UTC时间0点触发，可根据需求调整时区）
• 选择目标为你创建的Lambda函数，完成配置

内容的提问来源于stack exchange，提问作者Ho Jung Kim