如何使用Lambda+Python实现S3文件采集并发送至EC2的SFTP文件夹

1. 前置准备 #

• 确认EC2上的SFTP服务可正常访问，提前获取SFTP的访问地址、端口、登录账号、密码/密钥、目标写入路径，确认该账号对目标路径有写入权限
• Lambda函数需配置和EC2互通的网络：要么和EC2在同一个VPC下，要么配置VPC peering打通网络，同时EC2的安全组要开放SFTP端口（默认22）给Lambda的出口IP段
• 给Lambda配置IAM权限：至少包含S3桶的s3:GetObject权限、Lambda基础执行权限（用于输出CloudWatch日志）
• 提前准备SFTP依赖：在Amazon Linux 2环境下执行pip install paramiko -t python/，将生成的python目录压缩为zip包，上传为Lambda层，关联到你的Lambda函数即可调用SFTP相关能力
• 在Lambda的环境变量中配置SFTP_HOST、SFTP_PORT、SFTP_USER、SFTP_PASS、SFTP_REMOTE_PATH五个参数，避免敏感信息硬编码

2. 配置S3触发规则 #

• 进入需要触发同步的S3桶的「事件通知」页面，新建事件通知
• 事件类型选择所有对象创建事件，可按需配置前缀/后缀过滤规则（比如只同步upload/目录下的.pdf文件，就填前缀upload/、后缀.pdf）
• 事件目标选择Lambda函数，选中你后续创建的Lambda函数即可

3. Lambda Python代码实现 #

import boto3
import os
import paramiko

s3_client = boto3.client('s3')
# 读取环境变量配置
SFTP_CONF = {
    "host": os.environ['SFTP_HOST'],
    "port": int(os.environ['SFTP_PORT']),
    "user": os.environ['SFTP_USER'],
    "passwd": os.environ['SFTP_PASS'],
    "remote_path": os.environ['SFTP_REMOTE_PATH']
}

def lambda_handler(event, context):
    # 解析S3触发事件的文件信息
    s3_record = event['Records'][0]['s3']
    bucket = s3_record['bucket']['name']
    file_key = s3_record['object']['key']
    file_name = os.path.basename(file_key)
    tmp_file_path = f"/tmp/{file_name}"

    try:
        # 下载S3文件到Lambda临时目录
        s3_client.download_file(bucket, file_key, tmp_file_path)
        # 建立SFTP连接
        transport = paramiko.Transport((SFTP_CONF['host'], SFTP_CONF['port']))
        transport.connect(username=SFTP_CONF['user'], password=SFTP_CONF['passwd'])
        sftp = paramiko.SFTPClient.from_transport(transport)
        # 上传文件到SFTP目标路径
        target_path = f"{SFTP_CONF['remote_path'].rstrip('/')}/{file_name}"
        sftp.put(tmp_file_path, target_path)
        # 释放资源
        sftp.close()
        transport.close()
        os.remove(tmp_file_path)
        return {"statusCode": 200, "msg": f"文件{file_key}同步成功"}
    except Exception as err:
        print(f"同步失败，错误信息：{str(err)}")
        # 可在此处添加告警逻辑，比如发送SNS通知
        raise err

如果你用密钥认证SFTP，把transport.connect部分替换为密钥认证逻辑即可，密钥内容可以存在AWS Secrets Manager中，Lambda运行时拉取到临时目录使用即可，不要硬编码在代码中。
如果需要同步大文件，可调整Lambda的临时目录存储空间（最大支持10GB）、超时时间（最大15分钟），也可以改成分段上传下载逻辑避免内存占用过高。

4. 测试验证 #

手动上传测试文件到S3对应路径，查看Lambda的CloudWatch日志排查错误，登录EC2的SFTP目录确认文件正常落地即可。

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