KafkaSpout的延迟非常高

Question

我是apache storm和kafka的新手，作为POC的一部分，我正在尝试使用Kafka和apache storm来处理消息流。我正在使用来自https://github.com/apache/storm/tree/master/external/storm-kafka的storm-kafka源代码，我能够创建一个示例程序，使用KafkaSpout从kafka主题读取消息并将其输出到另一个kafka主题上。我有3个节点kafka（所有三个在同一服务器上运行）集群，并创建了8个分区的主题。我将KafkaSpout并行度设置为8，将bolt的并行度设置为8，尝试使用8个执行器和任务。我尝试在kafka级别，SpoutConfig级别和风暴级别设置了很多tunnig参数，但是我的整体延迟问题非常高。我需要消息处理程序，因此确实需要acking。 Storm群集有一个主管，动物园管理员有3个noed，它在kafka和storm之间共享。它运行在Red Hat Linux机器上，带有144MB RAM和16CPU。使用以下参数，我可以获得非常高的spout进程延迟大约40Sec，我需要达到50K msg / sec级别，请你帮我配置来实现它。我在各种网站上经历了很多帖子，尝试了大量的调优选项而没有结果。

Storm config
topology.receiver.buffer.size=16
topology.transfer.buffer.size=4096
topology.executor.receive.buffer.size=16384
topology.executor.send.buffer.size=16384
topology.spout.max.batch.size=65536
topology.max.spout.pending=10000
topology.acker.executors=20

Kafka config
fetch.size.bytes=1048576
socket.timeout.ms=10000
fetch.max.wait=10000
buffer.size.bytes=1048576

提前致谢。

Storm UI截图

Answer 1

您的拓扑有几个问题：

1. 你应该有与kafka分区相同数量的spout执行器
2. 您的拓扑无法足够快地处理元组。我很惊讶元组如何不会因超时而失败。对topology.max.spout.pending使用合理的值，我推荐150或 这只会防止超时，您的spouts会慢慢消耗元组，因为拓扑的其余部分无法处理它。
3. 您需要为螺栓添加更多执行程序，只有拓扑变得更快才能使更多执行单元发挥作用。执行程序和线程不是一回事，你需要在拓扑中放置更多的执行程序。您的单个执行器延迟为0,097，这意味着您的单个执行器每秒可处理大约10309个元组;这是为了达到每秒50k的目标，你需要至少有5个执行者。我相信，使用16个cpu机器，你可以使用1个以上的CPU来处理螺栓。
4. 这些任务的主要目的是将它们 - 重新平衡 - 推广到执行者;因此num tasks> = num executors。
5. 如果使用全局分组，则需要重新设计拓扑以使用字段分组等字符串。

Answer 2

查看您的UI屏幕截图，您的spouts似乎会发出更多数据，因为您的螺栓可以处理这些数据。两个喷口都发出大约500K消息，但只有250k消息（同样可以通过执行的螺栓元组的数量来推断 - 它大约是480K，这是两个喷口发出的元组的一半）。 40s的延迟是否从一开始就是相同的值？或者延迟会随着时间而增加？如果它随着时间的推移而增加，很明显你的螺栓是瓶颈。您有两种选择：

1. 增加螺栓和/或螺栓的平行度
2. 设置参数spout.max.pending以节流喷口输出率

如果你有足够的核心，那么第一个选项才有意义（但是到目前为止这不应该是一个问题，因为你提到了16个可用的CPU）。如果第二个选项适用于您，则取决于您要实现的吞吐量。你提到了50K msg / sec，但UI没有显示当前的吞吐量数字（即喷口输出率），因此我无法判断限制是否是一种选择。此外，你必须通过试错法确定spout.max.pending的最佳值（从1000的值开始对我来说似乎是合理的）。

Answer 3

我不知道您的问题是否已解决，但除了根据您的延迟要求调整topology.max.spout.pending之外，您还需要调整批量大小。将topology.spout.max.batch.size设置为较低的数字可能有助于减少延迟。