如何通过Lambda函数将S3中的多份CSV文件导入DynamoDB（JSON格式）

我来给你详细拆解怎么用Lambda实现这个需求，另外还有几个实用的替代方案，你可以根据自己的场景选～

一、用Lambda函数实现的完整步骤 #

整个流程是：S3上传CSV后自动触发Lambda → Lambda读取CSV内容转成JSON → 批量写入DynamoDB。具体操作如下：

1. 给Lambda配置必要的权限 #

Lambda得有三个关键权限才能干活：

• 读取目标S3桶的权限（s3:GetObject）
• 写入指定DynamoDB表的权限（dynamodb:BatchWriteItem）
• 生成日志的权限（方便排查问题，logs:CreateLogGroup、logs:CreateLogStream、logs:PutLogEvents）
  你可以去IAM控制台新建一个角色，把包含这些权限的策略附加给它，然后把这个角色分配给你的Lambda函数就行。

2. 设置S3触发Lambda的规则 #

打开S3控制台找到你的桶，进入「属性」→「事件通知」，新建一个通知：

• 事件类型选「所有对象创建事件」或者「Put」（看你是要触发所有上传还是仅特定操作）
• 可以设置前缀/后缀过滤，比如只对.csv后缀的文件触发
• 目标选Lambda函数，指定你要创建的那个函数

3. 编写Lambda函数代码（Python示例） #

Python处理CSV和JSON特别顺手，Lambda对Python的支持也很成熟，给你个参考代码：

import boto3
import csv
import json

# 初始化服务客户端
s3 = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
# 替换成你的DynamoDB表名
table = dynamodb.Table('your-dynamodb-table-name')

def lambda_handler(event, context):
    # 从S3事件里拿到文件信息
    record = event['Records'][0]
    bucket_name = record['s3']['bucket']['name']
    file_key = record['s3']['object']['key']
    
    # 读取S3里的CSV文件
    response = s3.get_object(Bucket=bucket_name, Key=file_key)
    csv_content = response['Body'].read().decode('utf-8').splitlines()
    
    # 把CSV解析成字典列表（也就是JSON格式的结构）
    csv_reader = csv.DictReader(csv_content)
    items = []
    for row in csv_reader:
        # 这里可以根据你的DynamoDB表结构调整字段，比如把字符串转成数字、处理日期等
        items.append({'PutRequest': {'Item': row}})
    
    # 批量写入DynamoDB，注意一次最多写25条，所以要拆分列表
    if items:
        for i in range(0, len(items), 25):
            batch = items[i:i+25]
            table.batch_writer().batch_write_item(RequestItems={table.name: batch})
    
    return {
        'statusCode': 200,
        'body': json.dumps(f'成功处理 {len(items)} 条数据')
    }

几个要注意的点：

• 确保CSV的表头和DynamoDB表的字段名对应，要是不一样，得在代码里做字段映射
• DynamoDB的主键字段必须存在，不然写入会失败
• 如果CSV文件很大，记得调整Lambda的超时时间（默认3秒，大文件可以调到5分钟）和内存（比如1024MB），避免中途挂掉

4. 测试验证 #

手动上传一个测试CSV到S3，然后去CloudWatch Logs看Lambda的运行日志，或者直接去DynamoDB表里检查数据有没有成功导入。

二、其他可行的解决方案 #

如果Lambda不是你的菜，还有这几个方案可以选：

1. AWS Glue（适合大规模数据ETL） #

Glue是专门的ETL服务，适合处理大量数据的场景：

• 先创建Glue爬虫，爬取S3里的CSV文件，自动生成表结构
• 再创建Glue作业，写个ETL脚本把CSV转成JSON，然后写入DynamoDB
• 可以设置定时触发，或者当S3有新文件时自动触发作业
  优点是能处理超大批量数据，不用自己管服务器；缺点是配置相对复杂一点，适合有一定AWS基础的用户。

2. Amazon Athena + DynamoDB（适合需要先过滤数据的场景） #

如果你需要先对CSV数据做查询过滤，再导入DynamoDB，可以用这个组合：

• 先在Athena里创建外部表，关联S3的CSV文件
• 然后用INSERT INTO语句把查询结果直接写入DynamoDB（要给Athena配好写入DynamoDB的权限）
  这种方式不用写代码，用SQL就能搞定，适合懂SQL的用户。

3. 本地脚本+AWS CLI（适合一次性需求） #

如果只是一次性处理，不需要自动化，那直接写个本地脚本就行：

• 用AWS CLI把S3的CSV文件下载到本地
• 用Python/Java等语言把CSV转成JSON
• 再用AWS CLI或者SDK把JSON数据批量写入DynamoDB
  优点是简单快速，不用配置各种服务；缺点是不能自动触发，每次都得手动跑脚本。

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