问题标签 [amazon-dynamodb-streams]

我有这个设置

ApiGateway -> Lambda1 -> DynamoDB -> Lambda2 -> SNS -> SQS

这是我想要做的事情：

1. 向 ApiGateway 发出 http 请求。
2. ApiGateway 集成到 Lambda1，因此 Lambda1 被执行。
3. Lambda1 将一个对象插入到 DynamoDB。
4. DynamoDBStream 触发 Lambda2。批量大小为 100。
5. Lambda2 为每条插入的记录发布一条消息到 SNS。
6. SQS 订阅了 SNS。

基本上，如果我向 Api Gateway 发出 http 请求，我希望看到一条消息以 SQS 结尾。实际上，对于单个请求，一切都按预期工作。

我做了这个测试：

1. 发出 10 个 http 请求来预热 lambda 函数并等待 30 秒。
2. 创建 100 个线程。每个线程将发出一个 http 请求，直到总请求数为 10000。

测试的第二步在 110 秒内完成。我的 DynamoDB 配置为每秒 100 次写入，这 110 秒非常合理。110 秒后，我在 DynamoDB 表中看到这 10000 条记录

问题是消息在 SQS 中结束需要太多时间。我检查了 Lambda2 的日志，发现它在测试完成后 30 分钟仍然被触发。同样在 Lambda2 的日志中，我看到了这种模式。

日志由这些行的重复组成。100 行“已发布消息”是有意义的，因为 DynamoDBStream 的批处理大小配置为 100。对 Lambda2 的每个请求需要 50-60 秒，这意味着所有消息需要大约 90 分钟才能最终到达 SQS。

困扰我的是，每个“开始请求”都在“结束请求”之后。因此，根本原因似乎是 DynamoDBStream 没有并行触发 Lambda2。

问题

为什么 DynamoDBStream 不并行触发 lambda 函数？我缺少配置吗？

解决方案

在从答案和评论中得到建议后，这是我的解决方案。

1. 在发布每条消息之前，我正在重新创建 SNS 客户端。我在课堂上将其设为静态变量，Lambda2 在大约 15 秒内开始执行。
2. 然后，我将 DynamoDB 触发器的批量大小增加到 1000。
3. 在 Lambda2 中，我使用 10 个并行线程处理（发布到 SNS）DynamoDB 记录。
4. 将 Lambda2 内存分配从 192MB 增加到 512MB。

通过这些优化，我可以在 SQS 中看到所有 10000 条消息，在发送所有 http 请求后 10-15 秒。

结论 ：）

为了找到最佳（便宜且可接受的延迟）解决方案，我们需要使用不同的批量大小、线程数、分配的内存等进行几次测试。

在 Java 中，使用 Jackson ObjectMapper，我试图反序列化从 dynamo db 流中读取的 dynamo db 对象。

我首先调用： record.getDynamodb().getNewImage().get("primaryKey").getS().toString() 从流中获取“1_12345”的primaryKey值。

然后，我在对象映射器中使用它来创建具有 primaryKey 成员集的 Metrics 对象的新实例：objectMapper.readValue("1_12345", Metrics.class);

问题是我在那个电话上遇到了一个例外： Unexpected character ('_' (code 95)): Expected space separating root-level values

Metrics.class 是一个没有构造函数的简单 POJO。我想知道在 readValue 调用中是否需要任何特殊注释或转义字符。在出现此错误的情况下，我似乎找不到任何明确的解决方案。

（旁注 - 我无法直接从 json 解析它的原因是因为从流中解析 json 时的结构并不简单，值看起来像这样，S 表示字符串，N 表示数字等 {primaryKey={S: 1_12345,}, rangeKey={N: xxx}... etc. }：）

dynamodb文档说有分片，需要先迭代它们，然后对于每个分片，需要获取记录数。

该文件还说：

（如果您使用 DynamoDB Streams Kinesis Adapter，这将为您处理：您的应用程序将以正确的顺序处理分片和流记录，并自动处理新的或过期的分片，以及在应用程序运行时拆分的分片。有关更多信息，请参阅使用 DynamoDB Streams Kinesis Adapter 处理流记录。）

好的，但是我使用 lambda 而不是 kinesis（它们彼此相关吗？），如果 lambda 函数附加到 dynamodb 流，我应该关心碎片吗？或者我应该只编写 labda 代码并期望 aws 环境只将一些记录传递给那个 lambda？

我从 github 克隆了 aws-sdk-cpp，并且成功地构建了它（测试也通过了）。我没有做“make install”。我想专门编译 SDK 的 dynamodbstreams 部分，所以我在 cmake 参数中添加了 -DBUILD_ONLY="dynamodbstreams" 。我写了一个小测试代码，但由于报告为未定义的 DescribeStreamOutcome 类型（“inclomplete 类型”）而无法编译。但是，它存在于它应该存在的头文件中。

cmake 版本：3.5.1，g++ 版本：5.4.0（目前我没有尝试过clang）

任何人都可以查看代码并指出问题所在吗？

我有以下代码：

有了这个 Makefile：

制作后出现以下错误：

但是，DescribeStreamOutcome 是在 DynamoDBStreamsClient.h 中定义的：

那你能帮帮我吗？谢谢。

我使用 Firebase 创建了一个聊天应用程序。但我计划将后端从谷歌移动到 AWS。我发现 AWS 网站中DynamoDB支持的Stream功能与 Firebase 非常相似。

http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html

如果有人添加新消息。Firebase 和 DynamoDB 将通知客户端更改。这是聊天应用程序的基本功能。看起来DynamoDB Stream应该是 Firebase 的替代品。

但我注意到文档和示例DynamoDB Stream非常罕见。最奇怪的是教程已被删除

https://aws.amazon.com/getting-started/projects/build-mobile-messaging-app-ios/?nc1=h_ls

点击Get Started将被重定向到MobileHub。

不知道DynamoDB Stream以后是不是不支持这个功能了？而这项服务将被MobileHub.

我MobileHub在另一个应用程序开发中使用了另一个功能。但是不知道怎么用MobileHub搭建聊天功能

如果我想用它AWS来构建一个聊天应用程序。什么是最好的解决方案？有什么有用的例子或教程DynamoDB Stream吗？非常感谢。

挑战

我想用作为智能缓存的 API 网关构建一个简单快速（< 50ms）的 API。对 API 的请求应该根据它们的密钥从 DynamoDB 中批量获取一些项目。但是，如果在 dynamoDB 中找不到一个或多个项目，我想以某种方式触发 lambda 函数，谁知道如何在 dynamoDB 中填充那些丢失的项目以供以后的请求。

我的想法

我这样做的方法是仅使用密钥在 DynamoDB 中创建任何缺失的项目，然后使用 DynamoDB 流调用 lambda 函数，该函数读取记录并用来自另一个外部 API 的值填充它们的空白属性。

问题

然而，我的问题是，据我所知，对于 API 网关的每个请求，我只能向 dynamoDB（BatchGetItem 或有条件的 PutItem）发出一个请求。有什么方法可以实现我想要的。

我真的很想使用 API 网关，因为它需要非常积极地扩展，我宁愿不处理服务器和应用程序的扩展。

备择方案

另一种选择是让客户端代码决定响应中缺少哪些项目，然后发送第二个请求以“排队”那些用于填充的项目，但是我真的希望该逻辑不要在客户端进行并且并且还避免了两次网络往返。

我还研究了 dynamoDB 是否能够为我执行这种“查找或创建”行为——但似乎没有运气：可以触发什么事件来触发 DynamoDB 中的 lambda 函数？

希望创建一个 DynamoDB 全局表来存储客户信息。我遇到的问题是我当前的模式是收听此表上的更改并使用 Lambda 触发器发送电子邮件更新。

即您的个人资料信息已更改。如果这不是你..

我现在是否需要在每个区域都拥有该 Lambda，并且数据复制是否意味着每个区域都会触发它？

我已经在 DynamoDB 中创建了一个包含近 20 亿行的表。

由于查询要求，我不得不在其中创建一个全局二级索引（GSI）。GSI 创建过程在 36 小时前开始，但仍未完成。Portal 显示项目数量约为 1 亿。这么长的路要走。

问题：

1. 为什么分配足够的 WCU 和 RCU 需要这么长时间（实际上是 30k）。
2. 我使用的 GSI 分区键的值是重复的，这可能是创建 GSI 需要更多时间的原因（理想的情况是我们选择一个不会重复的分区键以使项目跨越多个分区）。
3. 有没有办法在进程启动时中止 GSI 的创建？它不允许通过 AWS 控制台。

谢谢。

我正在创建无服务器框架项目。

DynamoDB 表由其他 CloudFormation 堆栈创建。

我如何引用现有的 dynamodb 表的 StreamArnserverless.yml

我有如下配置

我们是否可以不使用 logstash 直接将 dynamo db 数据流式传输到 AWS 弹性搜索服务，因为使用 logstash 会产生额外费用？在我在线阅读的所有文章中，我们可以使用 logstash 或 lambda 来实现这一点。