使用AWS Lambda运行Python脚本：如何存储及检索数据？

嘿，我来帮你理清把本地Python脚本迁移到AWS Lambda后的存储方案——毕竟本地的CSV和SQLite在无服务器环境里根本hold不住，得换成AWS原生的云服务才行！

Lambda是无服务器计算服务，它的执行环境是临时创建的，执行结束后就会销毁。虽然Lambda提供了/tmp临时目录（最大10GB），但这个目录里的文件会随着执行环境销毁而消失，而且同一函数的下次执行不一定能复用之前的环境，所以绝对不能依赖它来持久化数据。

根据你的数据类型和使用场景，这里有几个最常用的选择：

1. Amazon S3（存CSV/Parquet等文件） #

如果你的数据适合以文件形式存储（比如和本地CSV用法类似），S3是首选——它是AWS的对象存储服务，持久化、成本低、支持高并发访问，和Lambda集成也超简单。

实现步骤： #

• 给Lambda的执行角色添加S3读写权限（比如AmazonS3FullAccess，生产环境建议用更细粒度的权限，比如只允许访问指定桶）
• 用Python的boto3库把数据上传到S3，示例代码如下：

import boto3
import pandas as pd
from io import StringIO

def lambda_handler(event, context):
    # 先执行你的API采集逻辑，得到data_df（假设是Pandas DataFrame）
    data_df = pd.DataFrame(...)  # 替换成你的API采集代码
    
    # 把DataFrame转为CSV字节流（不用存到本地文件）
    csv_buffer = StringIO()
    data_df.to_csv(csv_buffer, index=False)
    
    # 初始化S3客户端，上传文件
    s3 = boto3.client('s3')
    s3.put_object(
        Bucket='your-bucket-name',  # 替换成你的S3桶名
        Key=f'data/raw_data_{pd.Timestamp.now().strftime("%Y%m%d_%H%M%S")}.csv',  # 用时间戳命名避免覆盖
        Body=csv_buffer.getvalue()
    )
    
    return {"statusCode": 200, "message": "数据已成功保存到S3"}

检索数据： #

同样用boto3的get_object方法下载文件到Lambda的/tmp目录，再读取处理：

def get_s3_data(bucket_name, file_key):
    s3 = boto3.client('s3')
    response = s3.get_object(Bucket=bucket_name, Key=file_key)
    csv_content = response['Body'].read().decode('utf-8')
    return pd.read_csv(StringIO(csv_content))

2. Amazon RDS/Aurora（替代SQLite的关系型数据库） #

如果你的数据需要复杂的SQL查询、事务处理，或者要和其他业务系统共享数据，RDS（比如MySQL、PostgreSQL）或者Aurora（兼容MySQL/PostgreSQL的高性能数据库）是完美替代SQLite的选择。

实现步骤： #

• 确保Lambda所在的VPC能访问RDS实例（生产环境建议把RDS放在VPC内，给Lambda配置VPC访问权限；测试环境可以临时开RDS公网访问，但不推荐生产用）
• 用AWS Secrets Manager存储数据库的用户名和密码，避免硬编码（给Lambda角色添加Secrets Manager的读取权限）
• 用对应的数据库驱动连接并写入数据，示例（PostgreSQL）：

import psycopg2
import boto3
from aws_secretsmanager_caching import SecretCache, SecretCacheConfig

def get_db_credentials(secret_name):
    # 从Secrets Manager获取数据库凭证
    client = boto3.client('secretsmanager')
    cache = SecretCache(config=SecretCacheConfig(), client=client)
    secret_str = cache.get_secret_string(secret_name)
    return eval(secret_str)  # 假设凭证是JSON格式的字典

def lambda_handler(event, context):
    creds = get_db_credentials('your-db-secret-name')
    # 建立数据库连接
    conn = psycopg2.connect(
        host=creds['host'],
        database=creds['dbname'],
        user=creds['username'],
        password=creds['password']
    )
    cur = conn.cursor()
    
    # 假设API返回的数据是列表形式的records
    records = [...]  # 替换成你的API采集数据
    # 批量插入数据
    insert_sql = "INSERT INTO your_table (col1, col2, col3) VALUES (%s, %s, %s)"
    cur.executemany(insert_sql, records)
    
    conn.commit()
    cur.close()
    conn.close()
    
    return {"statusCode": 200, "message": "数据已成功写入RDS"}

检索数据： #

直接执行SELECT查询即可，比如：

def query_rds_data():
    creds = get_db_credentials('your-db-secret-name')
    conn = psycopg2.connect(**creds)
    cur = conn.cursor()
    cur.execute("SELECT * FROM your_table WHERE create_time >= %s", ('2024-01-01',))
    rows = cur.fetchall()
    cur.close()
    conn.close()
    return rows

3. Amazon DynamoDB（NoSQL数据库） #

如果你的数据是非结构化/半结构化，或者需要高并发的读写操作，DynamoDB是最佳选择——它是完全托管的无服务器NoSQL数据库，和Lambda天生适配，不用管服务器运维。

实现步骤： #

• 给Lambda角色添加DynamoDB的读写权限
• 用boto3的DynamoDB资源客户端写入数据：

import boto3

def lambda_handler(event, context):
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('your-table-name')
    
    # 假设API返回的数据是字典列表
    items = [...]  # 替换成你的API采集数据
    # 批量写入数据（注意DynamoDB批量操作的大小限制）
    with table.batch_writer() as batch:
        for item in items:
            batch.put_item(Item=item)
    
    return {"statusCode": 200, "message": "数据已成功写入DynamoDB"}

检索数据： #

按主键查询用get_item，条件查询用query，全表扫描用scan（不推荐大表用scan）：

def get_dynamodb_item(item_id):
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('your-table-name')
    response = table.get_item(Key={'id': item_id})
    return response.get('Item')

如果只是处理过程中需要临时缓存数据，可以用Lambda的/tmp目录，最大支持10GB，但记住：

• 这个目录是临时的，函数执行结束后可能被销毁
• 同一执行环境复用的时候可能还存在，但绝对不能依赖这个特性
  示例：临时存文件再上传到S3

import pandas as pd
import boto3

def lambda_handler(event, context):
    data_df = pd.DataFrame(...)
    # 临时存到/tmp
    temp_file_path = '/tmp/temp_data.csv'
    data_df.to_csv(temp_file_path, index=False)
    
    # 上传到S3
    s3 = boto3.client('s3')
    s3.upload_file(temp_file_path, 'your-bucket-name', 'data/temp_data.csv')
    
    return {"statusCode": 200}

• 依赖包处理：如果你的脚本用到了pandas、requests这类第三方库，需要把依赖打包成Lambda Layer，或者用Docker容器镜像部署（避免部署时因为缺少依赖报错）
• 超时设置：Lambda默认超时3秒，如果你的API采集+存储耗时久，要在控制台把超时时间调长（最长15分钟）
• 内存配置：Lambda的内存大小会影响CPU和网络性能，根据你的脚本需求调整（比如数据量大的话可以调大内存）

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