Lambda写入CSV到S3出现MemoryError 如何优化内存占用

原实现的内存开销根因 #

现有代码三个设计直接导致内存占满10GB配额：

• 逐文件把所有转换后的字典全量存入dicts列表，Python单个字典的内存开销是原始CSV文本的3-5倍，100万条字典仅这部分就会吃掉3-4GB内存
• 把全量CSV内容先写入io.StringIO，最后调用f.getvalue()获取全量字符串上传，相当于在内存中额外拷贝了一份400MB的完整CSV，内存占用直接翻倍
• 读取源S3文件时把所有行预先存入lines1列表，相当于把完整源文件内容全量加载到Lambda内存
• 冗余的全量缓存列表dictsToSend跨文件攒所有记录，完全没有实际业务用途却持续占用内存

核心改造思路 #

彻底抛弃“全量攒内存再上传”的模式，全程采用流式处理：从S3源流逐行拉取源文件、逐行做列名转换、逐行写入小容量缓冲、边攒够一定大小就分片推送到S3，内存中永远不会出现完整的源文件、完整的字典列表、完整的CSV字符串。

第一步：将全量返回的转换函数改造为逐行产出的生成器 #

删掉原来把所有转换结果存入列表的逻辑，改为边读S3流、边逐行转换、边逐行yield，不做任何全量缓存：

def transform_file_stream(file_obj, mapper_json, file_id, mail_filename):
    NCOAFields = mapper_json['mappings']
    body = file_obj['Body']
    # 直接读取流的第一行作为CSV表头，不缓存所有行
    header_line = next(body.iter_lines()).decode('utf-8').replace('"', '').split(',')
    # 直接迭代S3流的剩余内容传给CSV解析器，不落地为内存列表
    csv_reader = csv.DictReader(
        (line.decode('utf-8', errors='ignore') for line in body.iter_lines()),
        fieldnames=header_line
    )
    # 逐行转换，直接产出，不存入全局列表
    for row in csv_reader:
        transformed = {}
        metadata = {}
        for src_key, src_val in row.items():
            if src_key in NCOAFields:
                transformed[NCOAFields[src_key]] = src_val
            else:
                metadata[src_key] = src_val
        # 嵌套字典转为字符串写入CSV，避免对象额外内存开销
        if metadata:
            transformed['metadata'] = str(metadata)
        transformed['individual_id'] = str(uuid.uuid4())
        transformed['mail_filename'] = mail_filename
        transformed['file_id'] = file_id
        yield transformed

改造收益：读取源文件时不再需要把400MB全量加载到内存，逐行处理的内存占用降到单条记录级别。

第二步：基于S3分块上传实现边写边传，不缓存全量CSV #

利用AWS S3的分片上传能力，用10MB大小的二进制缓冲代替原来的全量StringIO，每攒够10MB内容就直接上传为一个独立分片，全程内存中仅保留当前缓冲的少量内容，不存储完整CSV：

1. 每个待上传文件初始化分片上传任务
2. 从生成器拿到第一条转换后的数据，确定CSV表头，写入缓冲
3. 逐行读取生成器产出的记录，写入缓冲，当缓冲大小达到10MB时立刻上传当前分片，清空缓冲继续处理
4. 所有记录处理完成后，上传缓冲中剩余的零散内容为最后一个分片，合并所有分片为最终CSV文件
   对应核心代码逻辑：

# 初始化分片上传任务
multipart_res = s3.create_multipart_upload(
    Bucket='slm-test-bucket-transactional',
    Key=f"{jsonRoot}.csv"
)
upload_id = multipart_res['UploadId']
uploaded_parts = []
part_seq = 1
# 用10MB容量的二进制缓冲代替全量StringIO
write_buffer = io.BytesIO()

# 启动转换流
data_stream = transform_file_stream(file, mapperJSON, file_id, i)
# 获取第一条记录确定表头
try:
    first_row = next(data_stream)
except StopIteration:
    # 空文件直接写入空对象，取消分片上传即可
    s3.put_object(Bucket='slm-test-bucket-transactional', Key=f"{jsonRoot}.csv", Body=b'')
    s3.abort_multipart_upload(Bucket='slm-test-bucket-transactional', Key=f"{jsonRoot}.csv", UploadId=upload_id)
    continue

# 初始化CSV写入器
csv_writer = csv.DictWriter(write_buffer, fieldnames=first_row.keys())
csv_writer.writeheader()
csv_writer.writerow(first_row)

# 逐行处理剩余记录
for row in data_stream:
    csv_writer.writerow(row)
    # 缓冲达到10MB立刻上传分片
    if write_buffer.tell() > 10 * 1024 * 1024:
        write_buffer.seek(0)
        part_content = write_buffer.read()
        part_resp = s3.upload_part(
            Bucket='slm-test-bucket-transactional',
            Key=f"{jsonRoot}.csv",
            PartNumber=part_seq,
            UploadId=upload_id,
            Body=part_content
        )
        uploaded_parts.append({'PartNumber': part_seq, 'ETag': part_resp['ETag']})
        part_seq += 1
        # 清空缓冲，准备写下一批内容
        write_buffer.seek(0)
        write_buffer.truncate()

# 处理最后一段剩余缓冲内容
if write_buffer.tell() > 0:
    write_buffer.seek(0)
    part_content = write_buffer.read()
    part_resp = s3.upload_part(
        Bucket='slm-test-bucket-transactional',
        Key=f"{jsonRoot}.csv",
        PartNumber=part_seq,
        UploadId=upload_id,
        Body=part_content
    )
    uploaded_parts.append({'PartNumber': part_seq, 'ETag': part_resp['ETag']})

# 合并所有分片完成上传
s3.complete_multipart_upload(
    Bucket='slm-test-bucket-transactional',
    Key=f"{jsonRoot}.csv",
    UploadId=upload_id,
    MultipartUpload={'Parts': uploaded_parts}
)
write_buffer.close()

注意：S3要求单个分片最小为5MB，设置10MB的缓冲阈值是兼顾内存占用和上传效率的最优值，过小会导致分片数过多增加API调用开销，过大会增加内存占用。

第三步：清理冗余的全量缓存 #

直接删掉代码中没有实际业务用途的dictsToSend列表，不要跨文件攒所有转换后的记录——按现有业务逻辑每个源文件处理完就单独生成一个目标CSV，完全不需要聚合所有记录到同一个列表，否则哪怕单文件做了流式优化，跨文件攒的列表还是会把内存占满。

改造后的实际效果 #

按上述方案改造后，Lambda运行内存占用会从原来的10043MB降到200-300MB，哪怕把Lambda的内存配置从10GB降到1GB，也能稳定处理400MB、100万行的CSV文件，不会触发MemoryError：

• 全程没有全量加载的源文件副本
• 没有全量缓存的字典列表
• 没有全量缓存的CSV字符串
• 内存占用和处理的文件大小完全解耦，仅和缓冲阈值、单条记录大小相关

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