Apache Storm：使不可扩展的资源可扩展

Posted 2023-04-18

【中文标题】Apache Storm：使不可扩展的资源可扩展

【英文标题】：Apache Storm: Make an Unscalable Resource Scalable

【发布时间】：2016-03-16 12:01:38

【问题描述】：

我最近开始向大数据世界介绍自己并尝试使用 Apache Storm。我遇到了以下问题，想了很多如何解决它，但我所有的方法似乎都很幼稚。

技术

Apache Storm 0.9.3、Java 1.8.0_20

上下文

有一个大的 xml 文件 (~400MB) 需要逐行读取 (xml-file-spout)。然后，每个读取文件行都由一连串螺栓发出和处理。

它必须是有保证的消息处理（通过锚定发送...）

问题

就文件相当大（包含大约 200 亿行）而言，我使用扫描仪读取它，基于缓冲流而不是将整个文件加载到内存中。到现在为止还挺好。当处理过程中某处出现错误时，问题就出现了：xml-file-spout 本身死亡，或者存在一些内部问题......

Nimbus 将重新启动 spout，但整个处理过程从头开始； 这种方法根本无法扩展。

解决思路

解决第一个问题的最初想法是将当前状态保存在某处：分布式缓存、JMS 队列、本地磁盘文件。当一个 spout 打开时，它应该找到这样的存储，读取状态并从指定的文件行开始。这里我也想过将状态存储在Storm的Zookeeper中，但是不知道是否可以从spout中寻址Zookeeper（有没有这种能力）？您能否为此建议最佳做法？

对于问题 2，我考虑将初始文件分解为一组子文件并并行处理它们。这可以通过引入一个新的“中断”喷口来完成，每个文件都将由一个专用螺栓处理。在这种情况下，保证处理会引发大问题，因为如果出现错误，必须对包含失败行的子文件进行完全重新处理（spout 的 ack/fail 方法）......您能否建议最佳实践解决这个问题？

更新

好的，到目前为止我做了什么。

先决条件

以下拓扑有效，因为它的所有部分（spout 和 bolts）都是幂等的。

引入了一个单独的 spout，它读取文件行（一个接一个）并将它们发送到中间 ActiveMQ 队列（'file-line-queue'），以便能够轻松重播失败的文件行（请参阅下一步);

为“file-line-queue”队列创建了一个单独的 spout，它接收每个文件行并将其发送到后续的 bolts。就我使用保证消息处理而言，在任何螺栓失败的情况下，都会重新处理消息，如果螺栓链成功，则会确认相应的消息（CLIENT_ACKNOWLEDGE 模式）。

如果第一个（文件读取）spout 失败，则会抛出 RuntimeException，这会终止 spout。稍后，专门的主管重新启动 spout，重新读取 inout 文件。这会导致消息重复，但就一切都是幂等的而言，这不是问题。此外，这里值得考虑一个状态存储库以产生更少的重复......

新问题

为了使中间 JMS 更可靠，我添加了一个异常监听器，它为消费者和生产者恢复连接和会话。问题出在消费者身上：如果会话恢复并且我在螺栓处理过程中有一条未确认的 JMS 消息，则在成功处理后我需要确认它，但只要会话是新的，我就会收到'找不到相关 ID 的问题。

有人可以建议如何处理它吗？

【参考方案1】：

首先回答您的问题：

是的，您可以将状态存储在 Zookeeper 之类的地方，并使用 Apache Curator 之类的库来处理。 分解文件可能会有所帮助，但仍不能解决您必须管理状态的问题。

让我们在这里谈谈设计。 Storm 是为流式构建的，而不是为批处理构建的。在我看来，更适合批处理的 Hadoop 技术在这里会更好：MapReduce、Hive、Spark 等。

如果您打算使用storm，那么它将有助于将数据流式传输到更易于使用的地方。您可以将文件写入 Kafka 或队列，以帮助您解决管理状态、确认/失败和重试的问题。

【讨论】：

Hello Kit Menke。谢谢你的回答。我将使用一些中间件来锻炼一个体系结构，以解耦难以维护的文件数据源和易于重放的错误文件行处理。是的，第一个想法是使用JMS，但这里的问题是有可能使JMS 队列超载。我可以尝试使用 xml-file-spout（作为生产者）和 JMS 中间件（作为消费者）之间的 java 阻塞队列来阻止它，但是是否有任何最佳实践（可能不仅是 JMS ......）？

一个常见的模式是 kafka ->storm -> hbase。是的，超载队列是一个问题，但无论如何您都需要考虑这一点。写入队列的应用程序需要能够处理“队列已满”错误，并且您需要确保 Storm 可以处理负载。