AWS云端实时MQTT数据处理存储选型与架构问题咨询

一、DynamoDB vs S3：存储选型分析 #

• 选DynamoDB的场景：
  • Web应用需要低延迟的随机查询（比如按设备ID查最新数据、单条记录检索）
  • 数据是结构化/半结构化，需要频繁读写、实时聚合分析
  • 高并发场景，要求每秒处理数千条写入请求
• 选S3的场景：
  • 数据需要批量归档、低成本长期存储
  • 后续要做大数据分析（比如EMR、Athena）或机器学习处理
  • 数据是非结构化（比如大 payload 的MQTT消息）或不需要实时随机访问
• 最佳实践：两者结合
  • 用DynamoDB存储热点实时数据，支撑Web应用的低延迟查询
  • 用Lambda或Kinesis Firehose将历史数据同步到S3归档，兼顾批量分析需求

二、当前架构（IoT Core → Kinesis → Lambda → DynamoDB）的问题排查与优化 #

数据完整性验证 #

• 检查Kinesis控制台的分片消费状态：查看每个分片的IteratorAge指标，若数值持续增长，说明有数据堆积，未被完全处理
• 查看Lambda的CloudWatch Logs：搜索写入DynamoDB的错误日志（如ConditionalCheckFailedException、ProvisionedThroughputExceededException），确认是否有写入失败的情况
• 在MQTT消息中加入全局唯一ID，写入DynamoDB时使用ConditionExpression = attribute_not_exists(message_id)，避免重复写入，同时验证每条消息是否都被存储

处理速度变慢的原因及优化 #

1. Kinesis分片瓶颈
   • Kinesis每个分片每秒最多处理1MB/1000条记录，若数据量超过分片承载能力，会导致堆积。根据当前流量调整分片数，分片数决定了最大并发处理能力
2. Lambda并发与批量设置
   • Kinesis触发的Lambda默认每个分片对应一个实例，若分片数少，并发上不去。可调整Lambda的批量大小（比如设为100条/批），减少Lambda调用次数，提高处理效率
   • 检查Lambda的执行超时时间，若处理单批数据超时，会触发重试，反而拖慢整体速度。根据单批数据处理时间合理设置超时
3. DynamoDB写入限流
   • 若使用Provisioned模式，前150条可能用了预留容量，后续超过阈值会触发ProvisionedThroughputExceededException节流。解决方案：
     • 切换到On-Demand模式，自动适配流量变化
     • 开启Provisioned模式的自动扩缩容，根据读写流量自动调整容量
     • 优化主键设计，避免热点分区（比如不要用递增时间戳作为主键，可采用设备ID+时间戳的复合主键）
4. 架构优化建议
   • 若数据量持续增长，可在Kinesis和Lambda之间加入Kinesis Firehose，自动批量写入DynamoDB或S3，减少Lambda的压力
   • 为DynamoDB开启流功能，后续可同步数据到Elasticsearch或Redshift，支撑更复杂的分析需求

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