在 Apache Flink 中使用 DynamoDB 流答案

【问题标题】：Consume DynamoDB streams in Apache Flink在 Apache Flink 中使用 DynamoDB 流
【发布时间】：2019-08-12 04:30:53
【问题描述】：

有没有人尝试在 Apache Flink 中使用 DynamoDB 流？

Flink 有一个 Kinesis 消费者。但我正在寻找如何直接使用 Dynamo 流。

DataStream<String> kinesis = env.addSource(new FlinkKinesisConsumer<>(
    "kinesis_stream_name", new SimpleStringSchema(), consumerConfig));

我尝试了很多搜索，但没有找到任何东西。然而，在 Flink Jira 板上发现了一个未决请求。所以我猜这个选项还不可用？我有什么选择？

Allow FlinkKinesisConsumer to adapt for AWS DynamoDB Streams

【问题讨论】：

标签： amazon-dynamodb apache-flink amazon-dynamodb-streams

【解决方案1】：

2019 年更新答案

FlinkKinesisConsumer 连接器现在可以在实现此JIRA ticket 后处理 DynamoDB 流。

更新答案

Apache Flink 好像没有使用 DynamoDB 流连接器adapter，所以可以从 Kinesis 读取数据，但是不能从 DynamoDB 读取数据。

我认为一种选择可能是实现一个应用程序，该应用程序将数据从 DynamoDB 流写入 Kinesis，然后在 Apache Flink 中从 Kinesis 读取数据并进行处理。

另一种选择是为 Apache Flink 实施自定义 DynamoDB 连接器。您可以使用existing connector 作为起点。

您还可以查看Apache Spark Kinesis connector。但它似乎也有同样的问题。

原始答案

DynamoDB 有一个 Kinesis 适配器，允许您使用 Kinesis Client Library 使用 DynamoDB 更新流。使用 Kinesis 适配器是使用来自 DynamoDB 的更新的推荐方式（根据 AWS）。这将为您提供与直接使用 DynamoDB 流（也称为 DynamoDB 低级 API）相同的数据。

【讨论】：

您好 Ivan，关于如何使用此适配器的文档不足。我运行了示例代码，但这仍然没有让我知道如何在 flink 中添加源。我必须在 flink 中创建自定义源吗？ Flink 文档也没有多说。一个示例会很有帮助。
我正在考虑编写 lambda 函数来将 dynamodb 流写入 kinesis。从性能和成本的角度来看，不确定该解决方案有多好。每秒写入超过 500 次，调用 lambda 函数的次数将是一个很大的数字。这是一个好的选择吗？
每秒 500 次写入大约是每月 21M 次写入。如果您为此功能分配 128MB 并且每次执行将花费 0.2 秒，根据此计算，您每月将支付约 4 美元 (aws.amazon.com/lambda/pricing)
我做错了什么。每秒 500 次写入是每天 500 * 60 * 60* 24 ~ 43 M 次调用。所以费用是每天 8 美元，我有 3 张这样的桌子。因此，它将是每天 24 美元。如果 Lambda 函数被批量调用而不是每次插入会更好:)
好的，你的计算是正确的，是的，每天 8 美元。但请记住，您支付的是执行时间，而不是每次调用。它假设您的功能将工作 0.2 秒，如果它更快，您将支付更少的费用。我不认为它可以按批次执行。