Python Lambda从DynamoDB导出CSV到S3：代码优化及报错咨询

一、先解决NameError报错 #

报错核心原因有3个：

1. 类名拼写不一致：类定义是DynamoDBExtractLambda，但实例化时写成了DynamoDbExtractLambda（注意DB的大小写差异）。
2. 实例化位置错误：把实例化代码放在了类内部，应该移到类定义外部。
3. 缺失必要导入：LambdaApplication类未导入，如果不是必须依赖这个类，新手建议直接用简单的函数式写法，避免不必要的复杂度。

修正后的基础可运行代码（简化版） #

import csv
import io
import os
import boto3
from datetime import datetime

# 全局复用boto3客户端，Lambda冷启动后可重复使用，提升性能
dynamodb = boto3.client('dynamodb')
s3 = boto3.client('s3')

def lambda_handler(event, context):
    # 安全获取环境变量，不存在时用默认值或抛出明确错误
    env = os.getenv("ENV_PREFIX", "dev")
    dynamodb_table_name = f"{env}-dynamodb-table"
    bucket = f"{env}-s3-bucket"
    
    # 生成带秒级时间戳的CSV文件名，避免同一分钟多次触发重名
    current_date = datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
    key = f"{current_date}-dynamodb-export.csv"

    try:
        # 处理DynamoDB扫描分页（默认最多返回1MB数据，需循环获取全部）
        items = []
        response = dynamodb.scan(TableName=dynamodb_table_name)
        items.extend(response['Items'])
        
        while 'LastEvaluatedKey' in response:
            response = dynamodb.scan(
                TableName=dynamodb_table_name,
                ExclusiveStartKey=response['LastEvaluatedKey']
            )
            items.extend(response['Items'])

        # 定义CSV表头
        CSV_HEADERS = ["header1", "header2"]

        # 内存中生成CSV文件，避免Lambda本地文件系统限制
        csv_buffer = io.StringIO()
        csv_writer = csv.DictWriter(csv_buffer, fieldnames=CSV_HEADERS)
        csv_writer.writeheader()

        # 转换DynamoDB带类型标识的数据为CSV行
        for item in items:
            csv_row = {}
            for header in CSV_HEADERS:
                # 兼容字符串、数字、布尔等常见DynamoDB数据类型
                attr = item.get(header, {})
                csv_row[header] = attr.get('S', attr.get('N', attr.get('BOOL', '')))
            csv_writer.writerow(csv_row)

        # 上传CSV到S3，指定编码和内容类型
        s3.put_object(
            Bucket=bucket,
            Key=key,
            Body=csv_buffer.getvalue().encode('utf-8'),
            ContentType='text/csv; charset=utf-8'
        )

        return {
            'statusCode': 200,
            'body': f'数据已提取并存储到 s3://{bucket}/{key}'
        }

    except Exception as e:
        # 记录详细错误日志，Lambda控制台可查看
        print(f"错误详情: {str(e)}")
        return {
            'statusCode': 500,
            'body': f'提取DynamoDB数据并存储到S3时出错: {str(e)}'
        }

二、代码方向确认 #

你的核心思路完全正确：

• 通过环境变量动态配置表名和存储桶，符合云原生最佳实践
• 用内存缓冲区生成CSV，规避Lambda本地文件系统的限制
• 使用触发时的时间戳作为文件名，满足需求
• 基础的异常处理框架已搭建完成

三、优化建议 #

1. 分页处理：原代码仅获取第一页数据，修正后加入分页循环，确保能拉取全量数据。
2. 环境变量安全读取：用os.getenv替代直接索引os.environ，避免变量缺失时直接崩溃。
3. 文件名规范：添加秒级时间戳避免重名，后缀改为.csv更直观。
4. 数据类型兼容：扩展处理DynamoDB常见数据类型（数字、布尔值等），避免数据丢失。
5. S3元信息配置：上传时指定ContentType，让S3和浏览器正确识别文件类型。
6. 日志优化：如果使用AWS Lambda Powertools库，可替换print为结构化日志工具，便于排查问题。
7. 表头灵活配置：表头可通过环境变量或DynamoDB表结构动态获取，减少硬编码。

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