如何使用AWS Lambda将DataFrame写入DynamoDB？——CloudFormation部署的Python3.8 Lambda测试通过但无法写入数据的求助

我一眼就看到了核心问题——你的DynamoDB写入逻辑完全放在了Lambda的lambda_handler函数外部，这就是数据没写入的根本原因！

关键错误拆解 #

AWS Lambda的执行规则是：只有lambda_handler函数是触发时的执行入口，函数外部的代码只会在**冷启动阶段（第一次执行或长时间闲置后首次执行）**运行一次，而且此时content变量还没被定义（它是lambda_handler内部的局部变量），这不仅导致写入逻辑不会被正常触发，甚至可能在冷启动时抛出NameError。

另外还有几个需要优化的细节：

• 硬编码AWS密钥严重违反安全最佳实践，Lambda应该通过IAM角色获取DynamoDB访问权限，无需在代码中写入密钥。
• pandas、geopy这类第三方库不是Lambda默认环境自带的，你需要把它们打包成Lambda层或者包含在部署包中，否则运行时会出现导入错误。
• 缺少错误处理逻辑，API请求失败、DynamoDB写入失败时没有捕获异常，不利于排查问题。

修正后的代码 #

import boto3
import pyowm
import time
import json
import requests
from datetime import datetime, date, timedelta, timezone
import pandas as pd
from geopy.geocoders import Nominatim

def lambda_handler(event, context):
    # 初始化DynamoDB资源（依赖IAM角色权限，无需硬编码密钥）
    dynamodb = boto3.resource('dynamodb', region_name='eu-west-1')
    table = dynamodb.Table("Dev_Weather")
    
    api_key = "xxxxxxx" # 替换为你的API Key
    owm = pyowm.OWM(api_key)
    city = 'Berlin, DE'
    geolocator = Nominatim(user_agent='aerieous@myserver.com')
    location = geolocator.geocode(city)
    lat = location.latitude
    lon = location.longitude
    
    # 设置日期为昨天并转换为时间戳
    x = (datetime.now() - timedelta(days = 1 ))
    d = x.isoformat(' ', 'seconds')
    p = '%Y-%m-%d %H:%M:%S'
    dt = int(time.mktime(time.strptime(d,p)))
    
    # 请求天气API并捕获异常
    url = "https://api.openweathermap.org/data/2.5/onecall/timemachine?lat=%s&lon=%s&dt=%s&appid=%s&units=metric" % (lat, lon, dt, api_key)
    try:
        response = requests.get(url)
        response.raise_for_status() # 捕获HTTP请求错误
        data_history = json.loads(response.text)
    except Exception as e:
        print(f"API请求失败: {str(e)}")
        return {"statusCode": 500, "body": f"API请求失败: {str(e)}"}
    
    # 处理天气数据
    df_history2 = pd.json_normalize(data_history, record_path='hourly', meta=['lat', 'lon', 'timezone'], errors='ignore')
    df_history2['dt'] = pd.to_datetime(df_history2['dt'],unit='s').dt.strftime("%m/%d/%Y %H:%M:%S")
    df_history2 = df_history2.rename(columns={'dt': 'timestamp'})
    df_history2['uuid'] = df_history2[['timestamp','timezone']].agg('-'.join, axis=1)
    df_select_hist2 = df_history2[['uuid','lat','lon', 'timezone', 'timestamp', 'temp', 'feels_like', 'humidity', 'pressure']]
    df_select_hist2 = df_select_hist2.astype(str)
    content = df_select_hist2.to_dict('records')
    
    # 写入DynamoDB并捕获异常
    try:
        for item in content:
            table.put_item(
                Item={
                    'pk_id': item['uuid'],
                    'sk': item['timestamp'],
                    'gsi_1_pk': item['lat'],
                    'gsi_1_sk': item['lon'],
                    'gsi_2_pk': item['temp'],
                    'gsi_2_sk': item['feels_like'],
                    'humidity': item['humidity'],
                    'pressure': item['pressure'],
                    'timezone': item['timezone']
                }
            )
        print(f"成功写入{len(content)}条数据到DynamoDB")
        return {"statusCode": 200, "body": f"成功写入{len(content)}条数据到DynamoDB", "data": content}
    except Exception as e:
        print(f"DynamoDB写入失败: {str(e)}")
        return {"statusCode": 500, "body": f"DynamoDB写入失败: {str(e)}"}

关键修改说明 #

1. 将DynamoDB操作移到lambda_handler内部：确保每次函数触发时都会执行写入逻辑，并且能正常访问content变量。
2. 移除硬编码密钥：改为依赖Lambda的IAM角色，你需要给Lambda角色添加dynamodb:PutItem权限，这是AWS的安全最佳实践。
3. 添加异常处理：捕获API请求和DynamoDB操作的异常，方便排查问题，同时返回明确的状态码和错误信息。
4. 简化代码逻辑：直接使用item字典中的值，无需单独赋值变量，让代码更简洁易读。

额外注意事项 #

• 依赖包打包：Lambda默认环境没有pandas、geopy、pyowm这些库，你需要通过pip install -t . <库名>安装后打包成部署包，或者创建Lambda层来管理这些依赖。
• 测试验证：修改后重新部署Lambda，测试时查看CloudWatch日志确认无报错，同时检查DynamoDB表是否有数据写入。

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