使用aiodynamo异步查询DynamoDB过慢，求优化及错误排查

当前使用aiodynamo查询DynamoDB表，条件为partitionKey=symbol、sortKey=EffectiveTime（Unix时间格式）按日期范围筛选，结果逐行写入对应日期的JSON文件。当symbol数量达到数千级时，代码运行缓慢，以下是具体优化建议与排查方向：

核心问题分析 #

• 串行遍历所有symbol，每个symbol单独发起查询，数千个symbol会产生大量串行请求，完全没有利用异步框架的并发优势。
• 每条查询结果立刻执行flush()，频繁的IO操作会显著拖慢整体速度。
• 未对查询参数做针对性优化，可能导致分页请求次数过多。

优化建议 #

1. 并发执行symbol查询 #

利用asyncio的gather并发处理同一日期下的所有symbol查询，避免串行等待。由于文件写入是同步操作，需要用异步锁防止并发写入冲突。

2. 优化文件IO操作 #

• 减少flush()调用频率：积累一定数量的条目后再批量写入并flush，降低IO开销。
• 使用with语句管理文件，自动处理关闭逻辑，避免资源泄漏。

3. 调整DynamoDB查询参数 #

• 设置合理的Limit参数：增大单次查询返回的条目数，减少分页请求次数（默认Limit可能较小，可根据数据量调整为1000或更高）。
• 用ProjectionExpression指定需要的字段：只返回业务所需数据，减少网络传输量。
• 按需选择一致性：如果不需要强一致性，保持默认的最终一致性，提升查询速度。

4. 复用客户端资源 #

当前代码中HTTP客户端和DynamoDB客户端的创建逻辑是合理的（单次创建复用），无需重复初始化，避免不必要的开销。

重构后的示例代码 #

import asyncio
import orjson as json
from httpx import AsyncClient
from aiodynamo.client import Client
from aiodynamo.expressions import HashKey, RangeKey, ProjectionExpression
from aiodynamo.credentials import FileCredentials
from aiodynamo.http.httpx import HTTPX
import pathlib
import os

aws_creds_path = pathlib.Path(os.getenv('AWS_CREDENTIALS_FILE'))

async def query_and_write(symbol, min_key, max_key, table, file_handle, lock):
    # 指定仅需返回的字段，减少数据传输量
    query = table.query(
        key_condition=HashKey('symbol', symbol) & RangeKey('EffectiveTime').between(min_key, max_key),
        limit=1000,  # 增大单次查询条目数，根据实际情况调整
        projection_expression=ProjectionExpression("symbol", "EffectiveTime", "price")
    )
    batch = []
    async for item in query:
        batch.append(json.dumps(item) + b'\n')
        # 每积累100条批量写入一次
        if len(batch) >= 100:
            async with lock:
                file_handle.write(b''.join(batch))
                file_handle.flush()
            batch = []
    # 写入剩余的条目
    if batch:
        async with lock:
            file_handle.write(b''.join(batch))
            file_handle.flush()

async def get_table_items():
    async with AsyncClient() as h:
        client = Client(HTTPX(h), FileCredentials(path=aws_creds_path, profile_name='prod_profile'), region='eu-west-2')
        table = client.table('stock-prices')
        symbols = ['symbol1', 'symbol2', 'symbol3']  # 实际为数千个symbol
        for asof in ['2023-06-05','2023-06-06', '2023-06-07', '2023-06-08', '2023-06-09']:
            min_key = beginning_day_unix(asof)
            max_key = end_day_unix(asof)
            # 使用with语句自动管理文件生命周期
            with open(f'stock_prices_{asof}.json', 'wb') as json_file:
                lock = asyncio.Lock()
                # 并发执行所有symbol的查询任务
                tasks = [
                    query_and_write(symbol, min_key, max_key, table, json_file, lock)
                    for symbol in symbols
                ]
                await asyncio.gather(*tasks)

async def main():
    await get_table_items()

asyncio.run(main())

性能排查方向 #

• 检查DynamoDB吞吐量限制：查看CloudWatch中的ReadThrottleEvents指标，如果存在节流情况，需调整Provisioned Throughput或开启Auto Scaling。
• 验证查询范围准确性：确认beginning_day_unix和end_day_unix生成的时间范围是否正确，避免查询到超出目标日期的数据。
• 网络延迟排查：如果本地与eu-west-2区域网络延迟较高，可考虑使用AWS VPC端点或调整运行环境到同区域。
• 数据量分析：统计每个symbol每天返回的数据条目数，若数据量极大，可考虑按时间段拆分查询，进一步提升并发效率。

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