在 Storm Bolt 中为单个元组获取 Kafka 偏移量答案

【问题标题】：Fetching Kafka Offsets in Storm Bolt for individual tuple在 Storm Bolt 中为单个元组获取 Kafka 偏移量
【发布时间】：2016-08-30 07:17:19
【问题描述】：

用例

使用 Apache Storm 将 Kafka 消息持久化到 S3

到目前为止的故事

我尝试使用 secor(https://github.com/pinterest/secor)，效果很好，达到了目的。但是，对于经理来说，这可能是一种过度维护（他们说他永远是对的）
我们已经拥有 Apache Kafka-Apache Storm 稳定集群，因此计划利用该基础设施。

议程和问题

来自 Kafka 的消息将在 Storm bolt 中进行批处理，并以文件的形式写入本地磁盘
经过一定的时间间隔和/或大小标准后，它将上传到 S3
为了管理故障，每个 Bolt 都应该能够跟踪 Kafka 分区并在理想情况下按元组偏移，因为 Bolt 将随机分布在集群中。
Partition / Offsets 可以持久化到 Zookeeper，但首先如何从 Tuple 中获取它们？除了将它们转发到 Kafka Spout 之外还有其他方法吗？

【问题讨论】：

【解决方案1】：

将KafkaSpout 配置为org.apache.storm.kafka.StringMessageAndMetadataScheme，这会将偏移量和分区添加到Spouts 发出的值中

【讨论】：

【解决方案2】：

Kafka spout 已经在 zookeeper 中跟踪主题的偏移量，所以你不需要在 bolt 中实现这个逻辑。

Kafka spout 将发出元组，拓扑将跟踪它。当元组被螺栓确认时，它通过了。 Spout 将考虑交付的元组。在 emmiting tuples spout 后面会跟踪 zookeeper 中的当前偏移量，所以如果出现问题，您可以开始读取消息而不是从开始。

上述拓扑将保证至少交付一次。使用三叉戟拓扑，您可以保证只交付一次。在这两种情况下，请查看 topology.max.spout.pending 设置。正确设置它至关重要，因为您将使用批处理。

【讨论】：

是的。喷口肯定会跟踪偏移量。但是在写入文件时，需要跟踪“最后写入的偏移量”和“最后上传的偏移量”。假设根据 spout 的当前偏移量是 Coff。在任何给定时间，Coff >= 最后一个偏移量写入 >= 最后一个偏移量上传。为了跟踪最后两个，它并不总是与 Coff 相同。
你想得到什么消息传递保证？至少一次还是恰好一次？
至少一次应该没问题。
那么，如果您对至少一次保证没问题，为什么要跟踪“最后写入的偏移量”？
好的。现在让我们跟踪一下。解决剩下的问题呢？