如何解决AWS Lambda解压S3大型ZIP文件超时及效率低下问题？

第三方定期将约2GB、包含5万份70-100KB XML文件的ZIP文件上传至S3存储桶，需用Python编写AWS Lambda函数将其解压后存入同桶的其他位置。现有方案处理小文件正常，但处理大文件时触发15分钟超时；采用stream-unzip流处理方案（尚未写回S3）时，约每2.7秒处理1个文件，5万份文件需耗时约1.5天。

现有代码概要 #

import json
...
from stream_unzip import stream_unzip
...

s3 = boto3.client('s3')

def zipped_chunks(in_bucket_name, in_filepath, in_region='us-gov-west-1'):
    yield from boto3.client('s3', region_name=in_region).get_object(
        Bucket=in_bucket_name,
        Key=in_filepath
    )['Body'].iter_chunks(102400)

def lambda_handler(event, context):

...
    try:
        cur_datetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        print(cur_datetime + ":  BEGIN")
        filenum = 0
        for file_name, file_size, unzipped_chunks in stream_unzip(zipped_chunks(my_bucket_name, zip_filepath)):
            filenum = filenum + 1
            chunk_num = 0
            for chunk in unzipped_chunks:
                chunk_num = chunk_num + 1
                #print(chunk)
                
            if filenum % 1000 == 0:
                cur_datetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                print(cur_datetime + ":  File number " + str(filenum) + ":  " + str(file_name) + " # of chunks:  " + str(chunk_num))
                
        cur_datetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        print(cur_datetime + ":  END")
        
    except Exception as e:
        error_message = "Error extracting ZIP file:  " + str(type(e)) + ":  " + str(e)
        print(error_prefix + error_message)

运行日志示例 #

2024-09-13T16:31:20.085Z
INIT_START Runtime Version: python:3.11.v7 Runtime Version ARN: arn:aws-us-gov:lambda:us-gov-west-1::runtime:d67b56f062769b7151cf7b87f96b6114c15520d33a6e6964d49d508181d135c5
2024-09-13T16:31:20.755Z
START RequestId: 3529bf8c-306d-4bf0-a83e-e1a6b8cf9230 Version: $LATEST
2024-09-13T16:31:20.851Z
BUCKET: idemia-tx-dl-dev-processing
2024-09-13T16:31:20.851Z
FILEPATH: extract/2024-09-09/DL/001/TTXDPS_IDEMIA_CPSMAIL_20240909_001.zip
2024-09-13T16:31:20.851Z
2024-09-13 16:31:20: BEGIN
2024-09-13T16:37:19.388Z
2024-09-13 16:37:19: File number 1000: b'03033123_867447766148600.xml' # of chunks: 2
2024-09-13T16:43:18.203Z
2024-09-13 16:43:18: File number 2000: b'05983127_233907390302466.xml' # of chunks: 3

函数最终因15分钟超时终止，需解决：是否有更高效的处理方法？代码是否存在问题？是否应放弃使用Lambda？

一、现有代码的核心问题 #

• 单线程串行处理：逐个文件串行解压，5万份文件全部走单线程，这是耗时久的根本原因。
• S3客户端未复用：zipped_chunks函数每次新建S3客户端实例，频繁创建销毁会增加额外开销。
• Chunk大小不合理：100KB的chunk过小，会增加IO操作次数，拖慢流处理效率。
• 无并行上传逻辑：后续添加写回S3操作时，单文件单独调用put_object会产生大量API请求，进一步降低速度。

二、高效处理方案 #

1. 单Lambda并行化处理 #

利用concurrent.futures.ThreadPoolExecutor实现多线程并行解压+上传，同时配合Lambda资源配置优化：

import concurrent.futures
import datetime
import boto3
from stream_unzip import stream_unzip

# 全局复用S3客户端，避免重复创建
s3 = boto3.client('s3', region_name='us-gov-west-1')

def zipped_chunks(in_bucket_name, in_filepath):
    # 调整chunk大小为1MB，减少IO次数
    yield from s3.get_object(Bucket=in_bucket_name, Key=in_filepath)['Body'].iter_chunks(1024*1024)

def process_single_file(file_name, unzipped_chunks, target_bucket, target_prefix):
    # 处理文件名的bytes转字符串，拼接目标路径
    target_key = f"{target_prefix}{file_name.decode('utf-8')}"
    # 流式上传文件内容
    s3.put_object(
        Bucket=target_bucket,
        Key=target_key,
        Body=b''.join(unzipped_chunks)
    )

def lambda_handler(event, context):
    my_bucket_name = 'idemia-tx-dl-dev-processing'
    zip_filepath = event['Records'][0]['s3']['object']['key']
    target_prefix = 'unzipped/'

    try:
        print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}: BEGIN")
        filenum = 0
        # 根据Lambda内存配置调整线程数：1024MB内存建议10-15线程，2048MB建议20-30线程
        with concurrent.futures.ThreadPoolExecutor(max_workers=15) as executor:
            futures = []
            for file_name, _, unzipped_chunks in stream_unzip(zipped_chunks(my_bucket_name, zip_filepath)):
                filenum += 1
                # 提交任务到线程池
                futures.append(executor.submit(
                    process_single_file,
                    file_name,
                    unzipped_chunks,
                    my_bucket_name,
                    target_prefix
                ))
                # 每处理1000个文件清理已完成的任务，避免内存占用过高
                if filenum % 1000 == 0:
                    print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}: Processed {filenum} files")
                    for future in concurrent.futures.as_completed(futures[:1000]):
                        futures.remove(future)
            # 等待所有任务完成
            concurrent.futures.wait(futures)
        print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}: END")
    except Exception as e:
        print(f"Error extracting ZIP file: {type(e)}: {str(e)}")

2. 任务拆分突破单Lambda超时限制 #

如果并行化后仍无法在15分钟内完成，可采用SQS+多Lambda实例的架构：

• 第一步：用一个Lambda读取ZIP文件的目录，将每个待解压文件的元信息（ZIP路径、文件名）发送到SQS队列。
• 第二步：配置Lambda并发数，让多个实例同时消费SQS消息，每个实例负责从ZIP中解压单个文件并上传到S3。
• 这种方式可弹性扩展，彻底突破单个Lambda的时间限制。

3. Lambda配置优化 #

• 提升内存配置：Lambda的CPU和网络带宽随内存增加而提升，建议设置为1024MB或2048MB，能显著提升并行处理效率。
• 保留15分钟超时上限：这是Lambda的最大超时时间，足够覆盖大部分并行处理场景。

三、是否需要放弃Lambda？ #

不需要，Lambda完全适配该场景：

• 单Lambda并行化方案配合高内存配置，可将处理时间压缩至数小时内（远低于15分钟超时上限）。
• 任务拆分+SQS方案可进一步突破时间限制，同时利用Lambda的弹性伸缩能力降低成本。
• 只有当业务量持续暴增或需要复杂流式处理时，才考虑切换到ECS/EKS或EC2，但当前场景用Lambda+SQS组合足够高效且成本更低。

内容的提问来源于stack exchange，提问作者DR - Idemia