AWS Lambda CloudWatch Jobs

需要自动化定期任务如备份、邮件发送或数据处理？AWS Lambda结合CloudWatch Events/EventBridge是替代传统cron作业的完美无服务器解决方案。

Lambda定时作业概述#

AWS Lambda是无服务器计算服务，允许您运行代码而无需管理服务器。与调度服务结合使用时，您可以：

自动化任务: 备份、报告生成、数据清理
节省成本: 仅在代码执行时付费
自动扩展: 无需配置即可处理负载增加
高可用性: AWS确保可用性

[!NOTE] AWS目前推荐在新定时作业中使用EventBridge Scheduler而不是传统的CloudWatch Events。

方法1: EventBridge Scheduler（推荐）#

EventBridge Scheduler是现代调度服务，与CloudWatch Events相比具有增强功能。

EventBridge Scheduler的优势#

时区支持: 不仅限于UTC
灵活时间窗口: 在灵活时间范围内运行
内置重试: 失败时自动重试
一次性调度: 支持单次执行
死信队列: 更好的失败处理

步骤1: 创建Lambda函数#

import json
import logging
import boto3
from datetime import datetime

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    """定时作业的处理程序"""
    logger.info(f"Job triggered at: {datetime.utcnow()}")
    logger.info(f"Event: {json.dumps(event)}")
    
    try:
        # 您的作业处理逻辑
        result = process_scheduled_task(event)
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Job completed successfully',
                'result': result
            })
        }
    except Exception as e:
        logger.error(f"Job failed: {str(e)}")
        raise

def process_scheduled_task(event):
    """主要作业处理逻辑"""
    # 示例: 数据库清理、发送报告等
    logger.info("Processing scheduled task...")
    
    # 您的业务逻辑在这里
    return {"status": "success", "processed_items": 10}

python

步骤2: 创建IAM角色#

Lambda需要执行作业和写入日志的权限：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

json

策略权限：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:*",  # 如果需要DynamoDB访问
        "s3:*"         # 如果需要S3访问
      ],
      "Resource": "*"
    }
  ]
}

json

步骤3: 使用EventBridge Scheduler创建调度#

使用AWS控制台#

转到EventBridge Scheduler控制台
点击Create schedule
选择Recurring schedule
配置调度表达式：

# 每天UTC时间9:00 AM运行
cron(0 9 * * ? *)

# 每15分钟运行
rate(15 minutes)

# 每周一9:00 AM运行
cron(0 9 ? * MON *)

# 每月1日运行
cron(0 0 1 * ? *)

bash

选择Lambda function作为目标
配置时区（如果需要）：

# 9:00 AM Eastern Time
cron(0 9 ? * MON *)
timezone: America/New_York

bash

配置灵活时间窗口（可选）：

# 在调度时间后15分钟内运行
mode: FLEXIBLE
maximum_window_in_minutes: 15

bash

设置重试策略：

maximum_retry_attempts: 3
maximum_event_age_in_seconds: 3600

bash

使用AWS CLI#

aws scheduler create-schedule \
  --name daily-job \
  --schedule-expression 'cron(0 9 * * ? *)' \
  --schedule-expression-timezone 'UTC' \
  --target '{
    "Arn": "arn:aws:lambda:us-east-1:123456789012:function:my-function",
    "RoleArn": "arn:aws:iam::123456789012:role/scheduler-role"
  }' \
  --flexible-time-window '{
    "Mode": "FLEXIBLE",
    "MaximumWindowInMinutes": 15
  }'

bash

使用Terraform#

resource "aws_scheduler_schedule" "daily_job" {
  name       = "daily-job"
  group_name = "default"

  flexible_time_window {
    mode                      = "FLEXIBLE"
    maximum_window_in_minutes = 15
  }

  schedule_expression          = "cron(0 9 * * ? *)"
  schedule_expression_timezone = "UTC"

  target {
    arn      = aws_lambda_function.my_function.arn
    role_arn = aws_iam_role.scheduler_role.arn

    retry_policy {
      maximum_retry_attempts       = 3
      maximum_event_age_in_seconds = 3600
    }

    dead_letter_config {
      arn = aws_sqs_queue.dlq.arn
    }
  }
}

hcl

方法2: CloudWatch Events（传统）#

CloudWatch Events是传统方法，仍然支持但功能较少。

步骤1: 创建Lambda函数#

与Scheduler方法相同。

步骤2: 创建CloudWatch规则#

使用AWS控制台#

转到CloudWatch控制台
选择Events → Rules
点击Create rule
选择Schedule expression
输入cron表达式：

# 每天UTC时间9:00 AM
0 9 * * ? *

# 每5分钟
rate(5 minutes)

bash

选择Lambda function作为目标
配置权限（AWS将自动创建）

使用AWS CLI#

aws events put-rule \
  --name daily-lambda-rule \
  --schedule-expression 'cron(0 9 * * ? *)'

aws events put-targets \
  --rule daily-lambda-rule \
  --targets '{
    "Id": "1",
    "Arn": "arn:aws:lambda:us-east-1:123456789012:function:my-function"
  }'

aws lambda add-permission \
  --function-name my-function \
  --statement-id daily-lambda-rule \
  --action 'lambda:InvokeFunction' \
  --principal events.amazonaws.com \
  --source-arn arn:aws:events:us-east-1:123456789012:rule/daily-lambda-rule

bash

Cron表达式指南#

EventBridge使用6字段cron表达式：

cron(Minutes Hours Day-of-month Month Day-of-week Year)

text

Cron表达式示例#

# 每天UTC时间9:00 AM
cron(0 9 * * ? *)

# 每周一9:00 AM
cron(0 9 ? * MON *)

# 每月1日00:00
cron(0 0 1 * ? *)

# 每15分钟
rate(15 minutes)

# 每1小时
rate(1 hour)

# 周一到周五9:00 AM
cron(0 9 ? * MON-FRI *)

# 月的最后一天
cron(0 0 L * ? *)

# 每月的第一个周一
cron(0 0 ? * MON#1 *)

bash

字段值#

字段	值	特殊字符
分钟	0-59	, - * /
小时	0-23	, - * /
日	1-31, L, W	, - * / ? L W
月	1-12, JAN-DEC	, - * /
星期	1-7, SUN-SAT, ?, L, #	, - * / ? L #
年	1970-2199	, - * /

[!WARNING] 当日或星期被约束时使用?，不要同时使用两个字段。

最佳实践#

1. 幂等性#

确保处理程序是幂等的，以便安全重试：

def lambda_handler(event, context):
    # 检查作业是否已运行
    if is_job_already_processed(event):
        logger.info("Job already processed, skipping")
        return {"status": "skipped"}
    
    # 处理作业
    result = process_job(event)
    
    # 将作业标记为已处理
    mark_job_as_processed(event)
    
    return result

python

2. 死信队列（DLQ）#

设置DLQ以处理失败的调用：

import boto3
import json

sqs = boto3.client('sqs')

def send_to_dlq(error_message, event):
    """将失败的事件发送到DLQ"""
    sqs.send_message(
        QueueUrl='YOUR_DLQ_URL',
        MessageBody=json.dumps({
            'error': error_message,
            'event': event,
            'timestamp': str(datetime.utcnow())
        })
    )

python

3. 重叠执行预防#

使用DynamoDB锁防止重叠执行：

import boto3
from datetime import datetime, timedelta

dynamodb = boto3.resource('dynamodb')
lock_table = dynamodb.Table('job-locks')

def acquire_lock(job_id):
    """获取锁以防止重叠"""
    try:
        lock_table.put_item(
            Item={
                'job_id': job_id,
                'locked_at': datetime.utcnow().isoformat(),
                'expires_at': (datetime.utcnow() + timedelta(minutes=10)).isoformat()
            },
            ConditionExpression='attribute_not_exists(job_id)'
        )
        return True
    except Exception:
        return False  # 锁已存在

def release_lock(job_id):
    """完成后释放锁"""
    lock_table.delete_item(Key={'job_id': job_id})

python

4. 监控和日志#

使用CloudWatch Logs和Metrics：

import time
import logging

logger = logging.getLogger()

def lambda_handler(event, context):
    start_time = time.time()
    
    try:
        # 自定义指标
        logger.info("JOB_STARTED")
        
        result = process_job(event)
        
        # 记录持续时间
        duration = time.time() - start_time
        logger.info(f"JOB_COMPLETED duration={duration}")
        
        return result
    except Exception as e:
        logger.error(f"JOB_FAILED error={str(e)}")
        raise

python

5. 错误处理#

优雅地处理错误：

def lambda_handler(event, context):
    try:
        result = process_job(event)
        return {
            'statusCode': 200,
            'body': json.dumps(result)
        }
    except ValueError as e:
        logger.error(f"Validation error: {str(e)}")
        return {
            'statusCode': 400,
            'body': json.dumps({'error': str(e)})
        }
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")
        # 发送到DLQ或警报
        send_alert(str(e))
        raise

python

比较: EventBridge Scheduler vs CloudWatch Events#

功能	EventBridge Scheduler	CloudWatch Events
调度类型	cron + rate + one-time	cron + rate
时区支持	✅ 任意时区	❌ 仅UTC
灵活窗口	✅ 是	❌ 否
内置重试	✅ 是	❌ 否（需要DLQ）
成本	$1.00/M调用	免费（前5M/月）
用例	生产作业	简单定期作业

实践示例: 每日数据库清理#

import boto3
import logging
from datetime import datetime, timedelta

logger = logging.getLogger()
dynamodb = boto3.resource('dynamodb')

def lambda_handler(event, context):
    """清理30天前的记录"""
    table = dynamodb.Table('user-activity')
    
    cutoff_date = (datetime.utcnow() - timedelta(days=30)).isoformat()
    
    try:
        # 扫描并删除旧记录
        response = table.scan(
            FilterExpression='created_at < :cutoff',
            ExpressionAttributeValues={':cutoff': cutoff_date}
        )
        
        deleted_count = 0
        with table.batch_writer() as batch:
            for item in response['Items']:
                batch.delete_item(Key={'id': item['id']})
                deleted_count += 1
        
        logger.info(f"Deleted {deleted_count} old records")
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'deleted_count': deleted_count,
                'cutoff_date': cutoff_date
            })
        }
    except Exception as e:
        logger.error(f"Cleanup failed: {str(e)}")
        raise

python

使用EventBridge Scheduler调度：

# 每天UTC时间2:00 AM运行
cron(0 2 * * ? *)

bash

故障排除#

作业未运行#

检查调度表达式: 确保cron语法正确
验证IAM权限: Lambda具有必要权限
检查CloudWatch Logs: 查找错误消息
验证目标ARN: 确保Lambda函数ARN正确

作业运行但失败#

审查CloudWatch Logs: 查找错误消息
检查超时: Lambda可能超时
验证资源权限: 数据库、S3等访问权限
本地测试: 本地运行Lambda进行调试

重叠执行#

实现锁机制: 使用DynamoDB或Redis
使用预留并发性: 限制并发执行
添加幂等性: 确保处理程序是幂等的

清理资源#

不再需要时，清理资源：

# 删除调度
aws scheduler delete-schedule --name daily-job

# 删除CloudWatch规则
aws events delete-rule --name daily-lambda-rule

# 删除Lambda函数
aws lambda delete-function --function-name my-function

# 删除IAM角色
aws iam delete-role --role-name scheduler-role

bash

结论#

AWS Lambda结合EventBridge Scheduler/CloudWatch Events为定时作业提供了强大的解决方案：

EventBridge Scheduler: 具有时区支持、重试、灵活窗口的生产作业
CloudWatch Events: 具有成本效益的简单定期作业

关键要点：

生产环境使用EventBridge Scheduler
实现幂等性和错误处理
使用CloudWatch设置监控
对失败的调用使用DLQ
防止重叠执行

Lambda定时作业概述#

方法1: EventBridge Scheduler（推荐）#

EventBridge Scheduler的优势#

步骤1: 创建Lambda函数#

步骤2: 创建IAM角色#

步骤3: 使用EventBridge Scheduler创建调度#

使用AWS控制台#

使用AWS CLI#

使用Terraform#

方法2: CloudWatch Events（传统）#

步骤1: 创建Lambda函数#

步骤2: 创建CloudWatch规则#

使用AWS控制台#

使用AWS CLI#

Cron表达式指南#

Cron表达式示例#

字段值#

最佳实践#

1. 幂等性#

2. 死信队列（DLQ）#

3. 重叠执行预防#

4. 监控和日志#

5. 错误处理#

比较: EventBridge Scheduler vs CloudWatch Events#

实践示例: 每日数据库清理#

故障排除#

作业未运行#

作业运行但失败#

重叠执行#

清理资源#

结论#

参考文献#