问题标签 [amazon-kinesis-analytics]

我使用 Kinesis 数据流作为源，使用 elasticsearch 作为接收器。

我正在使用 Flink 作业稍微处理这些数据，然后将这些数据下沉到 elasticsearch。

在生产环境中，Kinesis 数据流每秒可以生成 50,000 个事件。处理数据以处理 500,000 个事件需要花费大量时间，大约需要 50 分钟左右的时间。

Elasticsearch 7.7版在基于 SSD 的存储上运行。

弹性搜索节点：2

碎片：5

副本：每个分片 1 个

刷新间隔：1 秒（默认）

我们正在使用 AWS opensearch elasticsearch。

有人可以建议导致这种延迟的原因吗？

在 AWS kinesis 数据分析应用程序中，集成了 Apache flink 应用程序。在 apache flink 仪表板中，任务管理器计数为 2，但需要在 apache flink 中再添加 2 个任务管理器以提高性能。

我们如何在 Apache flink 应用程序中添加任务管理器？

我需要使用 EventBridge 规则运行 KDA 应用程序吗？可能吗

我们目前有一个需要.jar文件才能运行的AWS Kinesis Data Analytics 应用程序。

我们已自动部署驻留在 S3 存储桶中的 .jar 文件。

我们的问题是，每当更新 .jar 文件时，我们都被迫重新启动 kinesis 应用程序以获取导致停机的新构建

有没有人有解决方法或其他方式来部署应用程序而不会导致停机？

我们正在尝试在 AWS Kinesis 上运行查询，并使用自动缩放。

我们发现，如果用户在 AWS Kinesis 配置中指定了 Parallelism 选项，AWS Kinesis 会将指定的并行度应用于所有算子。

我们还尝试在不指定任何并行度的情况下运行查询（将ConfigurationType 设置为 DEFAULT）。在这种情况下，Kinesis 只会启动 1 个具有 1 个插槽的任务管理器，即使这还不足以运行我们的查询。

所以我们想知道 AWS Kinesis Auto Scaling 是否支持为不同的算子设置不同的并行度？

谢谢

我使用 Kinesis 数据流作为源，使用 elasticsearch 作为接收器。在 AWS Kinesis Data 分析应用程序中运行 Flink 作业。

示例事件：

我正在从前端收集这些视频观看事件，而视频每 5 秒为一位用户播放一次。这些事件用于计算用户的观看时间。

假设如果一个用户正在观看视频，则每 5 秒从前端生成一次此事件，并将其摄取到 Kinesis 数据流中。因此，有 10,000 个用户观看视频，因此在一分钟内总共会生成 120,000 个事件。

为了处理120,000 个事件，我的 Flink 作业几乎需要大约 4 分钟的时间。这是相当长的一段时间。

那么如何才能提高工作的绩效呢？我需要在1 分钟内实现这一目标。

我的工作是这样的：

所以这项工作是做什么的，首先从 Kinesis Data 流接收事件，然后我通过这个流键入，userId然后我做一些videoDuration1 分钟，然后这些数据进入丰富功能，我从 Redis 读取一些数据并丰富这个事件，然后我下沉这个事件到elasticsearch。

我已经尝试增加工作的并行度，它为 1 个并行度提供了最佳性能，大约 4 分钟。如果我增加并行度，它会花费更多时间，这很奇怪。尝试使用 2、4、8、16 等。增加并行度应该可以加快处理速度，不是吗？

谁能帮助我在 Flink 工作中缺少什么或我做错了什么，我需要做什么才能在 1 分钟内加快这些事件的速度？

我正在尝试在 AWS KDA 上将 apache beam 与 Flink 一起使用。此管道从 kinesis 读取一些数据，经过简单的转换后，它会尝试使用以下策略将结果分组到一个分片中：

问题是，我面临很多 s3 减慢异常，并且在发生错误的同时只有 100 个临时文件：

我正在使用 Flink 1.11 处理 Kinesis 应用程序，但在启动我的应用程序时出现以下错误：

我在 Kinesis 和 4 KPU 中使用默认配置。

我是使用实时应用程序的新手。目前，我正在使用 AWS Kinesis/Flink 和 Scala 我有以下架构：

旧建筑

如您所见，我使用 CSVTableSource 使用 CSV 文件。不幸的是，CSV 文件对于 Flink 作业来说太大了。该文件每天更新​​，然后添加新行。所以，现在我在一个新架构中工作，我想用 CSV 替换 DynamoDB。

新架构

我的问题是：你推荐什么来使用 DynamoDB 表？

PD：我需要使用 DynamoDB 表和 Kinesis Data Stream 数据进行左外连接

我有一个 Flink-1.13 程序，它从包含具有不同模式的记录的 Kinesis 流中读取数据。

我的程序遍历流中包含的所有可能模式，过滤主数据帧并使用 StreamingFileSink.forBulkFormat将GenericRecord记录写入 S3。

看起来像：

我的问题是，当我通知 ToGenericRecordMap() 生成的类型时（否则它属于 kryo 序列化并出现另一个错误），我得到以下异常：

有什么问题吗？我错过了什么吗？谢谢你的帮助。