如何读取 Kinesis Data Stream 中最旧的未处理记录答案

【问题标题】：How to read the oldest unprocessed record in Kinesis Data Stream如何读取 Kinesis Data Stream 中最旧的未处理记录
【发布时间】：2018-06-01 08:36:58
【问题描述】：

我是 AWS 新手，需要一些指导。

我想处理最旧的未处理记录，但我似乎无法正确设置参数。

Current Architecture

对于分片迭代器：

我试过 TRIM_HORIZON，它给了我自开始。
我还尝试了 LATEST，它只给了我一条最新记录。

不确定这些额外的细节是否会有所帮助，但是......

我正在通过 AWS 控制台上的 Lambda 将自己的记录放入
我正在通过查看 CloudWatch 中的日志文件进行调试
我正在通过分片迭代器（TRIM_HORIZON 和 LATEST）获取记录
我的 getRecords 限制设置为 100

提前致谢！

【问题讨论】：

试试看这个：data-artisans.com/blog/turning-back-time-savepoints
它听起来您需要保留最后一个NextShardIterator 中的GetRecords response..."When you read repeatedly from a stream, use a GetShardIterator request to get the first shard iterator for use in your first GetRecords request and for subsequent reads use the shard iterator returned by the GetRecords request in NextShardIterator."...我在您的问题中遗漏了什么吗？
我从 kinesis 读取的 Lambda 函数只有在 kinesis 中有新活动时才会触发。因此，每次触发 lambda 函数时，我都会获得一个新的分片迭代器，因为我需要再次调用 GetShardIterator。我将尝试查看是否可以保留 NextShardIterator，但我试图避免保留它以保持 lambda 函数无状态。我还有其他方法可以做到这一点吗？非常感谢！

标签： amazon-web-services aws-lambda amazon-kinesis

【解决方案1】：

没有“最旧的未处理记录”，因为 Kinesis 不知道您已处理什么（例如，您可能已获取记录但未对它们执行任何操作）。

如果您使用 Kinesis，我强烈建议使用 Kinesis Client Library，它具有 checkpoints 的概念 - 这些本质上是 ShardIterator AFTER_SEQUENCE_NUMBER 之上的一个很好的包装器，转换为“最旧的未检查点记录” - 或尽可能接近“最旧的未处理记录”。

（你总是可以自己实现这个逻辑，但为什么不重用亚马逊已经为你完成的工作呢？）

【讨论】：

啊，是的。我在看KCL。但是，我仍然无法弄清楚如何使它与 Lambda 一起工作。也许我将不得不启动一个 EC2 实例。
是的，KCL 将需要一个主机来运行； EC2 将为您做到这一点。可能也更具成本效益，除非您的分片非常流量低（在这种情况下，有比 Kinesis 更具成本效益的解决方案）
感谢您的指导！最后一件事，我的分片流量很低。您所说的这些“比 Kinesis 更具成本效益的解决方案”是什么？
在它的核心，将 Kinesis 视为一个队列。这是一个队列，您可以用它做一些花哨的事情，尤其是在大容量时，但对于低容量，还有其他东西也可以充当“队列”（功能略有不同 - 根据您的需要选择）。 SQS 和 SNS 是更明显的基于 AWS 的选择。
啊。我会读完的。非常感谢所有做出贡献的人！ :)