发出大输出元组时如何使用 Spock 测试 Storm
所以我有以下测试: def "test execute(inputTuple) 方法发出一个包含成员 ObjectType 的 outputTuple使用 inputTuple 中的 entity_id 检索"() {给出:元组 inputTuple = Mock(Tuple);列表objectTypeIDsEmittedByPreviousBolt = new ArrayList();obje
..
apache-storm相关内容
Storm UI 拓扑不起作用
我们正在 LocalCluster 中执行 Storm 拓扑.Storm 拓扑运行良好并且能够连接 Storm UI (8090).但是 Storm UI 没有显示正在运行的拓扑信息. LocalCluster cluster = new LocalCluster(); 并提交如下: bin/storm jar bin/StormTest-0.0.1-SNAPSHOT.jar com.abz
..
有没有办法在风暴中应用多个分组?
我想对我的拓扑应用“字段分组"以及“本地或随机分组"，这样每个 spout 只将数据发送到本地螺栓，但也使用我文档中的字段来决定它应该使用哪些本地螺栓 因此，如果有两个工作进程，每个进程有 1 个 Kafka-Spout 和 2 个弹性搜索螺栓，则 local-or-shuffle 分组给我以下内容: 每个KS --->两个本地 ES 螺栓 fields-grouping 给了我以下内容
..
Storm 导致对 Ignite log4j 的依赖冲突
我尝试在 Storm 集群上运行一个 Storm 拓扑拓扑 jar 是由提供的创建的.在拓扑螺栓内部，我想从 MyIgniteCache 模块读取数据，但出现以下错误.我认为依赖项(:/usr/hdp/2.6.0.3-8/storm/lib/log4j-slf4j-impl-2.8.jar:/usr/hdp/2.6.0.3-8/storm/lib/log4j-core-Storm-core 的 2
..
更改 Apache Storm 拓扑的 Log4J 配置
我目前正在使用以下命令通过我的 Java 应用程序以编程方式提交 Storm 拓扑: Nimbus.Client 客户端 = NimbusClient.getConfiguredClient(stormConfigProvider.getStormConfig()).getClient();client.submitTopology(this.topologyID.toString(),Stor
..
杀死风暴拓扑后的资源清理
我们有一个与 MariaDB 数据库交互的风暴拓扑.我们的 Bolt 实现了 IRichBolt 接口并覆盖生命周期方法.我们在 prepare 方法中打开一个数据库连接，并在 cleanup 方法中关闭它.cleanup 方法文档说: 当 IBolt 将要关闭时调用.不保证会调用清理，因为主管在集群上 kill -9 的工作进程.保证调用清理的一个上下文是在本地模式下运行 Storm 时拓
..
Storm拓扑部署超时
我正在尝试在我的 Macbook Pro 上设置 Apache Storm (1.0.2)，但如果我尝试部署拓扑，显然会遇到超时问题.用户界面也挂了，吐出同样的异常. 3491 [main] INFO o.a.s.StormSubmitter - 为 MD5-digest 生成的 ZooKeeper 秘密负载:-8915636774701640550:-651075265796178588635
..
Apache Storm - 从 SPOUT 访问数据库 - 连接池
有一个喷口，每次滴答都会进入 Postgre 数据库并读取额外的一行.喷口代码如下所示: class RawDataLevelSpout 扩展 BaseRichSpout 实现 Serializable {私人整数计数器；SpoutOutputCollector 收集器；@覆盖public void declareOutputFields(OutputFieldsDeclarer 声明者) {声
..
在运行时部署流处理拓扑?
大家好， 我有一个要求，我需要重新提取一些旧数据.我们有一个多阶段管道，其来源是一个 Kafka 主题.一旦将记录输入其中，它就会运行一系列步骤(大约 10 个).每一步都会对推送到源主题的原始 JSON 对象进行按摩，然后推送到目标主题. 现在，有时，我们需要重新摄取旧数据并应用我上面描述的步骤的一个子集.我们打算将这些重新摄取记录推送到不同的主题，以免阻止通过的“实时"数据，这可能
..
Apache Storm Kafka Spout 滞后问题
我正在使用 Storm 1.1.2 和 Kafka 0.11 构建要在 Docker 容器中启动的 Java Spring 应用程序. 我的拓扑中的一切都按计划工作，但在 Kafka 的高负载下，Kafka 滞后随着时间的推移越来越多. 我的 KafkaSpoutConfig: KafkaSpoutConfigspoutConf =KafkaSpoutConfig.builder(
..
apache Storm 可靠性超时配置
我在 Linux Ubuntu 中部署了一个 nodejs->kafka>storm->Mongo.原本一切正常.然后我更改了 Storm worker 中的方法，这使得 Storm worker 处理消息的速度非常慢，每条消息大约 1 分钟，我注意到消息从 Storm 中一次又一次地发送.我恢复到原来的方法，一切都很好.(原始方法处理时间为每条消息 90ms). 我想这是 Storm 的可
..
使用“storm jar"远程部署 Storm 拓扑Windows 上的命令
我正在尝试在 Windows 上使用storm jar 命令将风暴拓扑部署到远程机器 storm jar jarName.jar mainClass 这是我的storm.yaml文件: storm.zookeeper.servers:- “主机名"nimbus.host: "主机名"nimbus.thrift.port: 6627 但是每当我尝试部署时，它总是尝试部署到我的本地机器上.
..
如何在不使用 maven 的情况下创建 jar 时排除某些 Jar 依赖项?
我正在开发一个核心 Java 项目.我正在编写一个 Apache Storm 拓扑，需要在将拓扑绑定到 jar 时排除风暴 jar.有没有办法在不使用 maven 的情况下做到这一点?我知道在 maven 中我们可以使用 provided 但我需要一个替代方案. PS:我正在使用 Eclipse. 解决方案 如果你使用 Maven 而不是 Gradle，并且你来这里是为了在构建中排除
..
无法运行“Hello World"Elastic-Search的Java客户端
我无法运行一个非常基本的程序来创建 Elastic-Search 的“Hello World"java 客户端. 文档非常简洁地说明了在这种情况下该怎么做. 这是我的代码: 查找.-f型./pom.xml./src/main/java/examples/EsRoutingNodeClient.java 两个文件如下所示. 包含代码的 Java 文件: 包示例；导入 org.
..
Apache Storm:Nimbus 未在端口 6627 上启动
启动 Nimbus 后，我在端口 6627 上看不到任何内容.我收到连接被拒绝错误.Nimbus 日志中抛出以下错误: 6899 [main] ERROR com.smarterme.intake.EmbeddedTopologyRunner - 拓扑提交失败.....org.apache.thrift7.transport.TTransportException:java.net.Connec
..
多个 Storm 拓扑之间的通信
我正在尝试在生产中部署多个 Storm 拓扑.我检查了文档，但找不到任何关于拓扑是否可以通过本机方法进行通信的参考.有没有人对如何实现这一点有任何建议? 简而言之，我很想知道元组是否可以跨拓扑发送. 感谢您的帮助！ 解决方案 从理论上讲，您可能会做到这一点.实际上，没有. 如果您想通过元组进行通信，您应该在一个拓扑中完成所有操作.拓扑组织计算而不是代码，所以你想一起计
..
Storm Nimbus 序列号是什么意思?
我已经编译了一个拓扑.在 nimbus 状态下，我有一些这样的数字序列 executor->start-time-secs 序列号是什么意思? 解决方案 按顺序 :executor->start-time-secs [2 2],[3 3],[1 1] - 代表节点.这些节点在 :executor->node+port{} 中分配，如下所示，由拓扑进一步使用 :executor
..
NotSerializableException org.neo4j.kernel.EmbeddedGraphDatabase
我正在使用 neo4j 创建图形，将 mongodb 中的数据作为文档.独立代码在没有风暴的情况下运行良好.但是在将它与 Storm 集成时，我得到了 - java.io.NotSerializableException: org.neo4j.kernel.EmbeddedGraphDatabase 例外.不知道我得到这个的确切原因.如果有人遇到此类问题，请告诉我如何解决. 解决
..
Storm 如何知道消息何时“完全处理"了?
(还有几个关于超时和 maxSpoutPending 的问题) 我在 Storm 文档中看到很多关于消息被完全处理的参考.但是我的 KafkaSpout 如何知道消息何时被完全处理? 希望它知道我的 bolt 的连接方式，所以当我的 Stream 中的最后一个 bolt 确认一个元组时，spout 知道我的消息何时被处理? 否则，我认为在超时期限到期后，将检查消息的确认状态，如果
..
java.lang.NoClassDefFoundError: org/yaml
我是使用 Storm 尝试提交 Storm-starter 的新手，但是当我 mvn 包 我遇到了那个错误 java.lang.NoClassDefFoundError: org/yaml/snakeyaml/constructor/BaseConstructor,编译:(word_count.clj:16:1)在 clojure.lang.Compiler$InvokeExpr.eval(
..