问题标签 [apache-storm]

我一直在阅读有关Storm的信息，并在使用 Storm-starter 中的示例。

我想我明白了这个概念，它非常适用于许多情况。我有一个我想做的测试项目来了解更多关于这方面的信息，但我想知道 Storm 是否真的适合这个。

我遇到的概念问题是“流”定义。似乎 Storms 将作为订阅流并实时处理它的魅力，但我并没有真正的流，而是我想要处理的有限数据集合。

我知道这有 hadoop，但我对 Storm 的实时功能以及编写 Storm 的 Nathan 在他的演讲中提到的其他有趣点很感兴趣。

所以我想知道，人们是否编写 Spout 来轮询非流 API，然后区分结果可能是为了模拟流？

第二个重点是，似乎 Storm 拓扑在中断之前永远不会完成处理，这同样不适用于我的情况。我希望我的拓扑知道，一旦我的有限源数据列表完成，处理就可以终止并可以发出最终结果。

那么，从 Storm 的角度来看，这一切都有意义吗？还是我看错了？如果是这样，对于这种实时并行计算需求，您提出了哪些替代方案？

谢谢！

我正在评估使用 Storm（实时框架，而不是 IDE）构建一个网络爬虫。我的 Spout 将传递一个 URI 流，必须在第一个 Bolt 中调用和检索，这是用 php 编写的。

来自 node.js 背景，我知道可以使用回调以非阻塞方式构建这样的功能。我的幼稚想法是，只为那个 Bolt 配置很多 Tasks，以便在一些 Taks 等待答案的同时充分利用资源。问题是，php 以为每个进程分配大量资源而臭名昭著，我不知道 Storm 是如何在内部管理这些资源的，以及这是否是个好主意。

在这种环境中是否可能出现类似 node.js 的行为？或者我是否必须为这个 Bolt 切换到另一种语言才能实现它（如果是，我该怎么做）？

问题的简短版本：如何在 Twitter Storm 中构建一个故障安全的字数统计程序（拓扑），即使发生故障也能产生准确的结果？这甚至可能吗？

长版：我正在研究 Twitter Storm 并试图了解它应该如何使用。我按照教程进行操作，发现它是一个非常简单的概念。但是本教程中概述的字数统计示例不是容错的（因为螺栓将一些数据保存在内存中）。但是，如果将事件重新提交到链的开头（当某些螺栓失败时会发生这种情况），则将相同的数据保存在后端数据库中会导致重复计算。

我是否应该将 Twitter Storm 视为生成部分准确结果的实时平台，并且仍然依赖 MapReduce 来获得准确的结果？

Storm 与 Hadoop 相比如何？Hadoop 似乎是开源大规模批处理的事实标准，Storm 比 hadoop 有什么优势吗？还是它们完全不同？

这是我的输入 spout 的代码片段，用于将元组发送到一个处理节点，以便在集群上进行流处理。问题是 BlockingQueue 正在抛出 InterruptedException 。

异常描述如下：---

java.lang.InterruptedException10930 [Thread-20] INFO backtype.storm.util - 异步循环中断！

而 nextTuple(InputStreamSpout.java:65 是 ------>

谢谢

我正在设计一个系统，该系统应该分析大量用户事务并生成汇总度量（例如趋势等）。该系统应该工作快速、健壮和可扩展。系统是基于java的（基于Linux）。

数据来自生成用户事务日志文件（基于 CSV）的系统。系统每分钟生成一个文件，每个文件包含不同用户的交易（按时间排序），每个文件可能包含数千个用户。

CSV 文件的示例数据结构：

10:30:01,用户 1,...
10:30:01,用户 1,...
10:30:02,用户 78,...
10:30:02,用户 2,...
10: 30:03，用户 1，...
10:30:04，用户 2，
... . .

我计划的系统应该处理文件并实时执行一些分析。它必须收集输入，将其发送到几个算法和其他系统，并将计算结果存储在数据库中。数据库不保存实际的输入记录，而只保存有关事务的高级聚合分析。例如趋势等。

我计划使用的第一个算法需要至少 10 条用户记录以实现最佳操作，如果 5 分钟后找不到 10 条记录，它应该使用任何可用的数据。

我想使用 Storm 来实现，但我更愿意尽可能将这个讨论留在设计级别。

系统组件列表：

1. 每分钟监视传入文件的任务。

2. 读取文件、解析文件并使其可用于其他系统组件和算法的任务。

3. 一个组件为用户缓冲 10 条记录（不超过 5 分钟），当收集到 10 条记录或 5 分钟过去时，是时候将数据发送到算法进行进一步处理了。由于要求为算法提供至少 10 条记录，我想到了使用 Storm Field Grouping（这意味着为同一用户调用相同的任务）并跟踪任务中 10 条用户记录的集合，当然我计划有几个这样的任务，每个处理一部分用户。

4. 还有其他组件可以处理单个事务，对于它们，我计划创建其他任务来接收每个被解析的事务（与其他任务并行）。

#3 我需要你的帮助。

设计这样一个组件的最佳实践是什么？很明显，它需要为每个用户维护 10 条记录的数据。键值映射可能会有所帮助，是在任务本身中管理映射还是使用分布式缓存更好？例如 Redis 一个键值存储（我以前从未使用过它）。

谢谢你的帮助

我正在组建一个小型 4 节点集群，我将在该集群上运行风暴。我有一些关于网络方面的问题。首先，所有计算机都配备了千兆以太网，但是我目前拥有的集线器只能达到 100 兆位。我应该升级我的集线器吗？或者性能增益可以忽略不计？其次，我在一些网站上读到集线器不是最好的硬件，而交换机更适合我的目的。我正在尝试使用 Storm 让一台机器从互联网上提取数据，然后将其传递给其他机器进行处理。交换机或集线器会更有用吗？感谢大家的帮助。

我正在编写一个基于 web.py 的 Web 服务，使用 Storm 作为 ORM 层从 MySQL 数据库中查询记录。Web 服务是通过 mod_wsgi 在 Linux 机器上使用 Apache2 部署的。create_database()当脚本使用storm的方法启动时，我创建了一个到MySQL数据库服务器的连接。这也是我创建Store对象的地方，该对象稍后用于在请求进入时执行查询。

在几个小时不活动后，store.find()抛出一个DisconnectionError: (2006, 'MySQL server has gone away'). 我并不感到惊讶，因为 Apache/mod_wsgi 重用 Python 进程而没有重新初始化它们很长一段时间，数据库连接被删除。我的问题是如何正确处理这个问题？

我尝试建立一种机制，通过向 MySQL 服务器发送一个重复的“SELECT 1”（每 300 秒）来保持与 MySQL 服务器的连接。不幸的是，这解决了我们的测试机器上的问题，但没有解决我们的演示部署（哎哟），而两者共享相同的 MySQL 配置（wait_timeout设置为 8 小时）。

我已经搜索了将风暴存储重新连接到数据库的解决方案，但没有找到任何复杂的解决方案。唯一的建议似乎是必须捕获异常，将其视为不一致，调用rollback()商店然后重试。但是，这意味着我要么必须包装整个Store类，要么一遍又一遍地实现相同的重试机制。有更好的解决方案还是我在这里完全错了？

更新：我添加了一个 web.py 处理器，如果捕获到异常，则通过重新创建风暴存储来优雅地处理断开连接错误，然后重试 Andrey 推荐的操作。然而，这是一个不完整且次优的解决方案，因为 (a) 存储被少数对象引用以供重用，这需要额外的机制来重新连接每个对象上的存储引用，以及 (b)，它不包括在数据库上执行写入时的事务处理（回滚）。但是，至少目前对于 store 上的所有读取操作来说这是一个可以接受的修复。

我有一个现有的 Avro RPC 客户端，可以将数据发送到 Avro RPC 服务器。Avro RPC 服务器当前将数据写入 HDFS（并且还执行其他操作）。我们正在将我们的服务器进程更改为基于 Storm。我希望找到一种简单的方法将我的数据导入 Storm，希望使用我现在拥有的 Avro RPC 消息。

我一直在寻找一种方法来做到这一点，到目前为止没有成功。Storm 有一个 RPC 模型，但它似乎仅限于传递字符串，这是我想避免的（为什么我首先去了 Avro）。Zeromq 可能是一种可能性，但对于我想要做的事情似乎有限。

有人可以为我建议一种优雅的方式来让我的 Avro RPC（基于模式的数据）进入 Storm 吗？

谢谢！！！！

我正在尝试使用Storm-contrib的 SimpleMongoBolt 。我下载了源码，进入storm-contrib-mongo目录，运行mvn package和mvn install。一切正常，IntelliJ 能够在编码时解决问题。但是，当我尝试构建我的项目时，它会尝试在外部存储库中为该库找到一个 pom。当它找不到它时，它就会失败。我需要做什么来解决这个问题？