如何用Python从DynamoDB筛选指定日期至今的交易数据？

核心修改步骤 #

1. 补全必要的导入：需要引入boto3（你的原代码遗漏了）和Attr类来构建筛选条件；
2. 定义筛选的起始日期，并生成当前时间的标准格式（需与表中CreatedAt字段格式匹配）；
3. 在scan方法中添加FilterExpression，限定CreatedAt的范围；
4. 处理DynamoDB的分页返回，避免只获取第一页数据。

修改后的完整代码 #

import pandas as pd
import boto3
from boto3.dynamodb.conditions import Attr
from datetime import datetime

# AWS配置
aws_access_key_id = '*****'
aws_secret_access_key = '*****'
region='****'

# 初始化DynamoDB资源
dynamodb = boto3.resource(
    'dynamodb',
    aws_access_key_id=aws_access_key_id,
    aws_secret_access_key=aws_secret_access_key,
    region_name=region
)

transactions_table = dynamodb.Table('transactions_table')

# 定义筛选起始日期（格式需与表中CreatedAt字段一致，示例为ISO 8601格式）
start_date = '2024-01-01T00:00:00Z'
# 获取当前UTC时间的ISO 8601格式字符串
current_time = datetime.utcnow().isoformat() + 'Z'

# 存储所有符合条件的数据
items = []

# 首次扫描，添加日期范围筛选
response = transactions_table.scan(
    FilterExpression=Attr('CreatedAt').between(start_date, current_time)
)
items.extend(response['Items'])

# 处理分页：循环获取剩余数据，直到没有更多结果
while 'LastEvaluatedKey' in response:
    response = transactions_table.scan(
        FilterExpression=Attr('CreatedAt').between(start_date, current_time),
        ExclusiveStartKey=response['LastEvaluatedKey']
    )
    items.extend(response['Items'])

# 转换为DataFrame并输出
df_transactions_table = pd.json_normalize(items)
print(df_transactions_table)

特殊情况处理 #

如果你的CreatedAt字段存储的是Unix时间戳（数值类型），需要将日期转换为对应数值：

from datetime import datetime

# 起始日期转Unix时间戳（秒）
start_timestamp = int(datetime.strptime('2024-01-01', '%Y-%m-%d').timestamp())
# 当前时间戳（秒）
current_timestamp = int(datetime.utcnow().timestamp())

# 筛选条件改为数值范围
FilterExpression=Attr('CreatedAt').between(start_timestamp, current_timestamp)

如果CreatedAt是表的排序键，建议使用query方法替代scan（效率更高）：

from boto3.dynamodb.conditions import Key

# 假设分区键为'UserId'，需替换为你的实际分区键
response = transactions_table.query(
    KeyConditionExpression=Key('UserId').eq('目标用户ID') & Key('CreatedAt').between(start_date, current_time)
)

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