我阅读了官方 AWS Kinesis Firehose 的文档，但没有提到如何处理重复事件。有没有人有这方面的经验？我google了一下有人用ElasticCache做过滤，是不是需要用AWS Lambda来封装这样的过滤逻辑？有没有像 firehose 这样的简单方法将数据摄取到 Redshift 中，同时具有“exactly once”语义？非常感谢！
【问题讨论】：
标签： duplicates amazon-redshift amazon-kinesis-firehose
您可以在 Kinesis Stream 的两侧进行复制。您可能会将相同的事件两次放入流中，并且您可能会被消费者读取两次事件。
如果您尝试将事件放入 Kinesis 流，则可能会发生生产者端，但由于某种原因您不确定它是否写入成功，并且您决定再次放入。如果您正在获取一批事件并开始处理它们，并且您在设法检查您的位置之前崩溃，并且下一个工作人员正在根据上一个检查点从 Kinesis 流中选择同一批事件，则可能会发生消费者端序列号。
在开始解决此问题之前，您应该评估此类重复的频率以及此类重复对业务的影响。并非每个系统都在处理不能容忍重复的金融交易。不过，如果您决定需要进行此类重复数据删除，解决此问题的常用方法是使用一些事件 ID 并跟踪您是否已经处理了该事件 ID。
带有 Redis 的 ElasticCache 是跟踪事件 ID 的好地方。每次拾取一个事件进行处理时，检查是否已经在 Redis 的哈希表中存在，如果找到，则跳过它，如果没有找到，则将其添加到表中（使用基于此类复制的可能时间窗口的一些 TTL）。
如果您选择使用 Kinesis Firehose（而不是 Kinesis Streams），您将无法再控制消费者应用程序，也无法实施此过程。因此，您要么希望在生产者端运行此类重复数据删除逻辑，改用 Kinesis Streams 并在 Lambda 或 KCL 中运行您自己的代码，要么选择 Redshift 中的重复数据删除功能（见下文）。
如果您对重复不太敏感，可以在 Redshift 中使用一些函数，例如 WINDOW 函数中的 COUNT DISTINCT 或 LAST_VALUE。
【讨论】：
不确定这是否是一个解决方案。但要处理重复项，您需要编写自己的 KCL。 Firehose 不能保证没有重复。一旦您拥有自己的 KCL 使用者来处理来自 Kinesis 日期流的数据，您就可以摆脱 Firehose。 如果您这样做，您可以关注链接的文章（完整披露，作者在此处），该文章在通过 KCL 消费者进行重复数据删除和处理后将事件存储到 S3。
通过查看事件的 ApproximateArrivalTimestamp，根据 Kinesis 数据流接收事件的分钟数对事件进行分组来存储事件。这允许我们始终将事件保存在相同的键前缀上，给定一批记录，无论它们何时被处理。例如Kinesis 在 2020/02/02/ 15:55 Hrs 收到的所有事件将存储在 /2020/02/02/15/55/*。因此，如果密钥在给定的分钟内已经存在，则意味着该批次已被处理并存储到 S3。
您可以实现自己的ISequenceStore，这将在您的情况下针对 Redshift 实现（在本文中，它是针对 S3 完成的）。阅读下面的全文。
https://www.nabin.dev/avoiding-duplicate-records-with-aws-kcl-and-s3
【讨论】：