Python实现Lambda每次触发时从S3文本文件按顺序读取下一行

核心逻辑 #

你遇到的核心卡点是Lambda为无状态服务，每次运行结束后临时环境会被销毁，无法通过本地变量存储上次读取的行号信息，必须将读取进度存储在外部持久化服务中。最轻量化的方案是和源文本文件同S3桶存储一个独立的进度文件，记录上次读取到的行索引。

实现步骤 #

• 提前准备S3文件：将待发送的内容按行存入content.txt上传到S3桶，同路径下新建last_read_line.txt，初始内容写入0（代表首次读取第0行，可根据你的索引习惯调整为1）
• Lambda执行流程按以下顺序编写：
  1. 调用S3 SDK读取进度文件内容，转换为整数得到上次读取的行号
  2. 读取源文本文件，按换行符分割为行数组
  3. 校验行号合法性：如果行号大于等于内容总行数，说明所有内容已发送完毕，可自定义终止逻辑或重置行号循环发送
  4. 取出当前行号对应的内容，附加到邮件中调用邮件服务发送
  5. 将行号+1后写入进度文件，覆盖原有内容更新进度
• 配置CloudWatch Event规则，按每日一次的频率触发Lambda即可

核心代码示例（Python） #

import boto3

# 初始化S3客户端
s3_client = boto3.client('s3')
# 替换为你的实际配置
S3_BUCKET = "你的S3桶名"
CONTENT_FILE_KEY = "待发送内容文件的S3路径"
PROGRESS_FILE_KEY = "进度文件的S3路径"

def lambda_handler(event, context):
    # 读取上次进度
    progress_res = s3_client.get_object(Bucket=S3_BUCKET, Key=PROGRESS_FILE_KEY)
    last_read_line = int(progress_res['Body'].read().decode('utf-8').strip())
    
    # 读取所有待发送内容
    content_res = s3_client.get_object(Bucket=S3_BUCKET, Key=CONTENT_FILE_KEY)
    all_lines = content_res['Body'].read().decode('utf-8').splitlines()
    
    # 校验是否已读完所有内容
    if last_read_line >= len(all_lines):
        # 此处可自定义读完后的逻辑，比如重置进度从第一行重新发送
        # last_read_line = 0
        raise Exception("所有内容已全部发送完成，任务终止")
    
    current_content = all_lines[last_read_line]
    # 此处补充你的邮件发送逻辑，调用AWS SES或其他第三方邮件服务均可
    
    # 更新进度
    new_progress = str(last_read_line + 1)
    s3_client.put_object(
        Bucket=S3_BUCKET,
        Key=PROGRESS_FILE_KEY,
        Body=new_progress.encode('utf-8')
    )

注意事项 #

• 若源文本文件体积较大（超过100MB），每次全量读取会浪费Lambda资源，可提前预存每行的字节偏移量索引，直接按偏移量读取对应行，无需全量加载文件
• 建议给S3桶开启版本控制，避免进度文件误写导致的读取顺序错误
• 需给Lambda执行角色配置对应S3桶的读写权限，以及邮件服务的发送权限

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