寻求支持长时处理的AWS Lambda替代服务（S3触发Dragen场景）

一、适配长时处理的AWS替代服务 #

针对你1小时的Dragen任务需求，以下几个服务可以完美替代Lambda，避免超时限制：

• AWS Step Functions
  支持编排多步异步任务，整体执行时长上限可达1年。可以把原Lambda里的流程拆成独立步骤：检查S3文件是否凑齐、启动EC2实例、发送SSM命令、等待命令执行完成、停止EC2。Step Functions会自动处理等待逻辑，不用让单个计算资源一直挂着，完全适配长时任务场景。

• AWS Batch
  专门为批量计算任务设计的托管服务。可以把Dragen命令打包成容器镜像，S3上传事件触发Batch任务后，Batch会按需创建/销毁EC2实例，自动调度资源执行任务，任务超时时间可配置为小时级，完全覆盖你的处理需求，还不用手动维护实例。

• 异步化Lambda流程（无需换服务）
  不用替换Lambda，只需要把同步等待的逻辑拆成两个独立Lambda：

  1. 第一个Lambda：响应S3上传事件，检查文件数量，启动EC2并发送SSM命令，记录任务信息后立即返回（不会超时）。
  2. 第二个Lambda：通过CloudWatch Events监听SSM命令的状态变化（成功/失败），触发后获取执行结果并停止EC2。

二、现有代码的优化改造示例 #

改造后的第一个Lambda（仅触发任务） #

import boto3
from datetime import datetime

def lambda_handler(event, context):
    bucket = event['Records'][0]['s3']['bucket']['name']
    target_prefix = 'input-samples/'
    
    # 检查是否凑齐两个目标文件
    required_files = ['test_02.fastq.gz', 'test_03.fastq.gz']
    s3_client = boto3.client('s3')
    existing_files = []
    try:
        response = s3_client.list_objects_v2(Bucket=bucket, Prefix=target_prefix)
        existing_files = [obj['Key'].replace(target_prefix, '') for obj in response['Contents']]
    except Exception as e:
        print(f"Failed to list bucket contents: {str(e)}")
        return
    
    if not all(f in existing_files for f in required_files):
        print("Missing required input files, aborting")
        return
    
    ec2_client = boto3.client('ec2')
    ssm_client = boto3.client('ssm')
    dynamodb = boto3.client('dynamodb')
    
    instance_id = 'i-0ca693219ebbd0a07'
    # 检查实例状态，未运行则启动
    instance_state = ec2_client.describe_instances(InstanceIds=[instance_id])['Reservations'][0]['Instances'][0]['State']['Name']
    if instance_state != 'running':
        ec2_client.start_instances(InstanceIds=[instance_id])
        print(f"Starting EC2 instance {instance_id}")
    
    # 构建Dragen命令
    bash_command = f'/opt/edico/bin/dragen -f -1 s3://{bucket}/{target_prefix}test_02.fastq.gz -2 s3://{bucket}/{target_prefix}test_03.fastq.gz --ref-dir=/home/centos/hg19 --output-directory s3://priyanshu-test-exome/output --enable-map-align=true --RGID=\"A.A.1\" --RGSM=CNTRL-0000131 --output-file-prefix=\"samples\" --intermediate-results-dir=/ephemeral/intermediate-output'
    
    # 发送SSM命令
    ssm_response = ssm_client.send_command(
        InstanceIds=[instance_id],
        DocumentName='AWS-RunShellScript',
        Parameters={'commands': [bash_command]},
        CloudWatchOutputConfig={
            'CloudWatchLogGroupName': '/aws/lambda/solutions-team-dragen-lambda',
            'CloudWatchOutputEnabled': True
        }
    )
    command_id = ssm_response['Command']['CommandId']
    
    # 记录任务信息到DynamoDB，供后续Lambda使用
    dynamodb.put_item(
        TableName='DragenTaskRecords',
        Item={
            'CommandId': {'S': command_id},
            'InstanceId': {'S': instance_id},
            'Bucket': {'S': bucket},
            'CreatedAt': {'S': datetime.utcnow().isoformat()}
        }
    )
    
    print(f"Dragen task triggered, Command ID: {command_id}")
    return {"status": "task_started", "command_id": command_id}

处理任务完成的第二个Lambda #

import boto3

def lambda_handler(event, context):
    # 从CloudWatch Event获取任务状态信息
    command_id = event['detail']['command-id']
    instance_id = event['detail']['instance-id']
    task_status = event['detail']['status']
    
    ec2_client = boto3.client('ec2')
    ssm_client = boto3.client('ssm')
    dynamodb = boto3.client('dynamodb')
    
    # 处理任务结果
    if task_status == 'Success':
        output = ssm_client.get_command_invocation(CommandId=command_id, InstanceId=instance_id)
        print(f"Task succeeded: {output['StandardOutputContent']}")
    else:
        print(f"Task failed with status: {task_status}")
    
    # 停止EC2实例
    ec2_client.stop_instances(InstanceIds=[instance_id])
    print(f"Stopped EC2 instance {instance_id}")
    
    # 清理DynamoDB中的任务记录
    dynamodb.delete_item(
        TableName='DragenTaskRecords',
        Key={'CommandId': {'S': command_id}}
    )
    
    return {"status": "task_processed", "instance_stopped": True}

配套配置 #

需要创建CloudWatch Events规则：

• 事件源选择AWS Systems Manager，事件类型为Command Invocation State Change
• 添加过滤条件，只监听Success和Failed状态的命令
• 触发目标设置为第二个Lambda

三、各方案对比 #

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