如何通过AWS Glue、API Gateway、Lambda将S3数据写入DynamoDB？解决KeyError问题

问题背景 #

已创建AWS Glue作业从S3桶读取CSV文件并转换为JSON格式，通过API Gateway端点将数据发送至Lambda函数，期望将数据写入DynamoDB表，但Lambda函数执行时出现KeyError: 'data'错误。

错误原因 #

Lambda代码对API Gateway传递的event结构理解错误：

• Glue发送的是单条JSON格式记录，但Lambda代码错误尝试访问event['data']['headers']，而API Gateway实际会将POST请求的JSON请求体放在event['body']字段中。
• Glue代码中使用requests.post的data参数传递字符串，会导致请求以表单形式发送，API Gateway解析后的event结构不符合预期。

修复步骤 #

1. 调整Glue Spark代码 #

将请求发送方式改为JSON格式传递，确保API Gateway正确解析请求体：

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
import requests
import json

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

s3_input = "s3://dynamodb-bucket-******/raw/"
dynamic_frame = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": [s3_input]},
    format="csv",
    format_options={"withHeader":True}
)
df = dynamic_frame.toDF()
json_data = df.toJSON().collect()

api_url = "https://*****.execute-api.us-east-1.amazonaws.com/dev"
for record in json_data:
    headers = {'Content-Type':'application/json'}
    # 将JSON字符串转为字典，用json参数发送，确保请求体为JSON格式
    response = requests.post(api_url, json=json.loads(record), headers=headers)
    if response.status_code != 200:
        print(f"Error sending data: {response.content}")

job.commit()

2. 调整Lambda函数代码 #

正确解析API Gateway传递的event结构，提取请求体数据并写入DynamoDB：

import json
import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('my_dynamodb_table')

def lambda_handler(event, context):
    # 解析API Gateway传递的请求体
    data = json.loads(event['body'])
    # 写入DynamoDB（确保data包含表的主键字段）
    table.put_item(Item=data)
    return {
        'statusCode': 200,
        'body': json.dumps('Data inserted successfully')
    }

额外注意事项 #

• 确保DynamoDB表my_dynamodb_table的主键字段存在于每条CSV记录中，否则put_item会报错。
• 检查API Gateway的集成设置，确保选择了Lambda Proxy Integration（默认模式），这样请求体才会被正确封装到event['body']中。

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