我有一个拓扑结构，看起来如下：
注意，Bolt2和Bolt3接收来自Bolt1和Bolt4的元组。所有的螺栓都是运行python脚本的ShellBolts，ShellSpout是运行从RabbitMQ读取的python脚本的ShellSpout。除了Bolt4之外，一切都按预期工作。如果我一次向RabbitMQ添加一条消息，那么所有消息都会正常工作并干净地完成。如果我在Bolt4上挂起的消息排队，Bolt4将永远不会处理它。其他螺栓仍然执行它们的功能，但Bolt4只会坐在那里完成第一个。
one显示，所有元组都由bolt4执行，但只有一个元组被加进。没有人失败。我使用的是Storm0
浏览 2提问于2015-12-02得票数 0
回答已采纳
假设我们有以下拓扑
spout A -> bolt B -> bolt C -> bolt E
而bolt 是最后一个，它将信息保存在数据库中，因此不需要发出任何元组。如何实现这样的解决方案，
如果我没有定义output_fields，那么我会得到异常
Exception in thread "main" java.io.IOException: org.apache.storm.thrift.protocol.TProtocolException: Required field 'output_fields' is unset! Struct
浏览 0提问于2017-03-26得票数 0
我正在使用Parse.ly/streamparse ()编写一个风暴拓扑，其中包含一个从json文档发出的元组，该元组将针对指定字段进行解析，并保存每个指定字段中实例的运行总数，以及每个字段中最常见的十个实例。它工作得很好，除了进程在读完输入json之前被终止。我对此仍然是个新手，所以我的第一个方法是简单地检查我的storm拓扑中的每个文件，看起来是否有可能导致进程提前终止的代码，但我找不到这样的代码。谷歌也一直让我失望。
一些我觉得可能相关的文件(拓扑中的文件太多了，我几乎不能发布)
喷嘴：
from __future__ import absolute_import, print_func
浏览 1提问于2015-05-19得票数 2
我正在尝试更改storm中的wordcount示例:我希望将结果发送到kafka集群而不是显示结果。这是构建拓扑的代码：
Config config = new Config();
config.setDebug(true);
config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
//set producer properties.
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9
浏览 0提问于2017-11-03得票数 0
在mongodb中使用Apache Storm 2.1.0时，我在Bolt执行中出现错误： java.lang.NullPointerException
at org.apache.storm.mongodb.common.mapper.SimpleMongoLookupMapper.toTuple(SimpleMongoLookupMapper.java:46)
at org.apache.storm.mongodb.bolt.MongoLookupBolt.execute(MongoLookupBolt.java:70)
at org.apache.storm.e
浏览 20提问于2020-11-02得票数 0
我有一个kafkaspout，2个用于处理数据的螺栓，2个用于在mongodb存储处理数据的螺栓。
我正在使用apache流量来创建拓扑，在这里，我将数据读入kafka的喷口。一切运行良好，但每次我运行拓扑，它就会从一开始就处理卡夫卡中的所有消息。一旦它处理了所有的msg，它就不会等待更多的msg和崩溃。
如何使风暴拓扑只处理最新的消息.
这是我的拓扑文件.yaml
name: "kafka-topology"
components:
# MongoDB mapper
- id: "block-mapper"
className:
浏览 0提问于2018-05-26得票数 0
回答已采纳
我需要在Storm中使用node-client编写一个弹性搜索螺栓。
我不想使用Transport-client，因为它的效率较低(涉及节点之间的额外一跳)。
所以我试着用下面的代码构建我自己的ES螺栓：
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
Settings settings = Settings.settingsBuilder()
.put("http.enabled", false)
.put("disc
浏览 5提问于2016-02-27得票数 0
我正在使用ApacheStorm0.10.0-beta1，并开始将一些拓扑转换为Flux。我决定从一个简单的拓扑开始，从一个Kafka队列中读取并写入一个不同的Kafka队列。我得到了这个错误，我有一个困难的时间找出什么是错的。拓扑yaml文件遵循错误。
Parsing file: /Users/frank/src/mapper/mapper.yaml
388 [main] INFO o.a.s.f.p.FluxParser - loading YAML from input stream...
391 [main] INFO o.a.s.f.p.FluxParser - Not pe
浏览 4提问于2015-08-20得票数 0
回答已采纳
我一直在为我们的实验室项目使用Storm进行测试。当我使用Storm bolt执行一些计算任务时，我发现在Storm外部运行计算所用的时间大约是Storm bolt execute()方法内部时间的十分之一。
我还在一个螺栓的execute()方法中尝试了这一小段代码：
Random rd = new Random(System.currentTimeMillis());
double count =rd.nextDouble();
for( int i=0;i <50000000;i++ ) {
count *= rd.nextDouble();
}
上面的代码在Storm内
浏览 0提问于2017-08-08得票数 0
1回答
当Storm作业提交到Hadoop群集以使用hdfsbolt写入hdfs时，不会在Storm UI中创建拓扑。显示该错误的原因是code(org.apache.storm.hdfs.bolt.AbstractHdfsBolt.cleanup(AbstractHdfsBolt.java:261) ~f083f1dc515311e9868bcf07babd3298.jar:?)中使用了某些包。
错误：
42608 [Thread-20-bolt-executor[3 3]] INFO o.a.s.util - Async loop interrupted!
42608 [Thread-19-di
浏览 3提问于2019-03-29得票数 0
我需要在LocalCluster模式下运行多个storm项目。
ProjectA->TopologyA->Spout1(listing to some stream X )->Bolt1->Bolt2
ProjectB->TopologyB->Spout1(listing to some stream Y )->Bolt1->Bolt2
ProjectC->TopologyC->Spout1(listing to some stream Z )->Bolt1->Bolt2
当我运行项目A、B和C时，不知何故，流被其他项目的
浏览 4提问于2018-04-26得票数 0
我正在尝试使用"storm-hdfs连接器0.1.3“将数据写入HDFS。github URL：，我已经将这个依赖项添加到我的maven项目中。
<dependency>
<groupId>com.github.ptgoetz</groupId>
<artifactId>storm-hdfs</artifactId>
<version>0.1.3-SNAPSHOT</version>
<scope>provided</scope>
浏览 3提问于2014-11-03得票数 5
我从克隆了Storm源代码的最新版本(主版)。我正在使用Ubuntu 14.02。
但是当我运行"mvn test“命令时，测试进程会失败，并在But .storm.multilang- test中终止。
下面是test.xml.storm.multilang-test.xml文件的上下文：
171293 [Thread-1212-1] ERROR b.s.util - Async loop died!
java.lang.RuntimeException: backtype.storm.multilang.NoOutputException: Pipe to subprocess s
浏览 0提问于2015-08-03得票数 0
回答已采纳
假设我有一个spout，它将一个主题中的消息发送到两个流( stream1和stream2)中，两个螺栓消费来自这两个流的消息(bolt1 -> stream1，bolt2 -> stream2).Now如果Bolt2无法确认元组，消息是否只会被重放到Bolt2?根据storm (apache-storm-1.0.2)中的新消费者API，实现了一个重试机制，其中如果numFail> maxRetry被storm.My确认，那么元组被storm.My确认是否怀疑以这种方式确认后消息会发生什么？它会从kafka队列本身中删除吗？如果是这样，我们将丢失一些消息，其中有failed.
浏览 0提问于2016-11-02得票数 0
I am using Leiningen 2.5.3 on Java 1.7.0_79 OpenJDK 64-Bit Server VM.
我想用lein uberjar排除风暴核心罐。下面是我的project.clj
(defproject kafka2hdfs "0.1.0-SNAPSHOT"
:description "FIXME: write description"
:url "http://example.com/FIXME"
:license {:name "Eclipse Public License"
浏览 3提问于2015-11-02得票数 1
回答已采纳
我正在尝试创建一个混合的storm拓扑，它使用基于Java的spout和基于python的bolt。
对于基于python的螺栓，我编写了一个Java包装器：
class PythonBolt extends ShellBolt implements IRichBolt {
public PythonBolt() {
super("python", "C:\\somepath\\sample.py");
}
@Override
public void declareOutputFields(OutputField
浏览 17提问于2018-03-05得票数 0
回答已采纳
1回答
我正在运行一个3节点的Storm集群，.We提交了一个包含10个工作人员的拓扑，下面是拓扑的详细信息
拓扑详细信息：
kafkaspout: 1 (with 10 kafka partions and 10 executors)
no of bolt: 4 (with bolt1 - 30 executors,bolt2 - 50 executors, bolt3 - 80 executors and bolt4 has 100 executors)
kafka spout pending : 500 messages
我们正在处理每个day.Issue的8到1000万个数据，Toplolg
浏览 0提问于2018-02-28得票数 0
1回答
我正在使用Apache Storm，我想知道有没有可能用另一个这样的螺栓组成一个螺栓：
public class MyNewBolt extends BaseRichBolt {
private MyOldBolt bolt; // MyOldBolt also extends BaseRichBolt
public MyNewBolt(MyOldBolt bolt) {
this.bolt = bolt;
}
@Override
public void prepare(Map map, TopologyContext topolo
浏览 0提问于2014-06-09得票数 2
我有一个python类，它从Apache 项目继承而来。
我的班级如下所示：
import storm
class MyClassName(Storm.Bolt):
def initialize(self,conf,context):
self._conf = conf;
self._context = context
def process(self, in_tuple):
storm.ack(in_tuple)
if __name__ == '__main__':
MyClassName().run()
我将python文件(myfi
浏览 0提问于2018-02-06得票数 1
回答已采纳
这是我试图运行的基本字数拓扑。但是我收到的错误是'INFO org.apache.storm.zookeeper.server.SessionTrackerImpl - SessionTrackerImpl exited loop!‘。有人能帮我吗？？
当我删除cluster.shutdown()时，tweet接连不断，直到我按下cntrl+c。
import java.util.Arrays;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import bac
浏览 0提问于2017-04-19得票数 0
Spout -> Bolt1 -> Bolt2 -> Bolt3 -> Ack orginalID (释放队列中的消息)
基本拓扑似乎工作了几分钟，但在某些情况下，消息既不会发送也不会被使用，而且也永远不会调用螺栓。我不知道java线程是不是死了，因为日志中没有异常，也没有任何错误的迹象。
对我来说，Apache storm .92或.95都有相同的结果。
我编写storm程序已经有几年了，从来没有见过这个……
浏览 1提问于2015-09-04得票数 1
在运行时，我在MainTopology类中遇到了一个错误，它在第24行显示了一个我无法理解的错误。下面是主类: package streamPostsCount；
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
public class MainTopology {
public static void main(S
浏览 1提问于2016-03-21得票数 0
1回答
我正在尝试调试非常简单的拓扑(1口1螺栓)
public class JoinerTopologyTest {
public static void main(String[] args) throws IOException {
Config conf = new Config();
conf.setNumWorkers(5);
conf.setDebug(true);
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("SPOUT-1",new My
浏览 1提问于2016-07-14得票数 2
回答已采纳
在线系统，storm Bolt获取NullPointerException，虽然我想我在61行之前检查过它；它偶尔获取NullPointerException；
import ***.KeyUtils;
import ***.redis.PipelineHelper;
import ***.redis.PipelinedCacheClusterClient;
import **.redis.R2mClusterClient;
import org.apache.commons.lang3.StringUtils;
import org.apache.storm.task.OutputColle
浏览 0提问于2017-10-17得票数 0
Version Info:
"org.apache.storm" % "storm-core" % "1.2.1"
"org.apache.storm" % "storm-kafka-client" % "1.2.1"
我正在创建和试验我已经创建的Storm拓扑，它有4个螺栓和一个kafka管口。
我试着调优配置，比如这些螺栓的并行度，max-spout-pending，等等，看看我能从中得到多大的规模。经过一些配置后，config/result如下所示：
max-spout-pe
浏览 0提问于2018-10-27得票数 1
我有一个单管和3个螺栓(A，B，C)的拓扑。
Bolt A从spout接收元组，并通过确认它来发出新值。
螺栓B和螺栓C消耗螺栓A的元组，但没有连接螺栓B和C。你可以说是两个独立的螺栓。
如果Bolt B失败，Bolt C成功确认元组，storm是否会一次又一次地向Bolt C重放相同的元组？
我想要确保两个都成功，如果没有，元组应该只去失败的螺栓，而不是成功的。我如何才能做到这一点。
我试图通过嘲笑tuple在螺栓B中失败和在C中成功来查看，但storm将tuple重播到所有螺栓。
浏览 0提问于2016-04-20得票数 0
我们有一个用例，我们需要使用apache storm从kafka读取消息头，并将其传递给下游的bolt。在storm文档中，没有提到如何将kafka头部传递到storm拓扑。有没有人知道该怎么做？
浏览 20提问于2021-01-21得票数 0
我正在尝试写一个基于Storm的代码，它从一个主题读取消息，然后写回另一个主题。输入主题具有ProtoBuf格式的数据，输出将采用JSON格式。我无法做到这一点。
这是构建拓扑的代码：
Config conf = new Config();
//set producer properties.
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9093");
props.put("request.requir
浏览 2提问于2016-08-31得票数 2
1回答
风暴创建拓扑
、、
我正在尝试使用Eclipse在Linux中运行storm starter示例。我收到以下错误，并且从未调用过nexttuple函数。
错误：
35979 [main] INFO o.a.s.d.supervisor - Shutting down supervisor cfba8fc6- 81e6-47cb-b8b9-ec7c700f4dfe
35981 [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2000] WARN o.a.s.s.o.a.z.s.NIOServerCnxn - caught end of stream exception
org.
浏览 2提问于2016-04-14得票数 0
我们正在尝试使用Apache Storm来处理大量(假的)消息。消息示例：
"{"clientName":"Sergey Bakulin","sum":12925,"group":"property","suspicious":false,"clientId":2,"dt":1404387303764,"coord":{"lat":55.767842588357645,"lon":37.4692036182333
浏览 2提问于2014-07-03得票数 8
在使用Storm 1.1.0和我的拓扑时，我遇到了一个问题，当拓扑有一个具有高延迟的Bolt时，Storm会重新调度或让螺栓和喷嘴崩溃。
现在我创建了一个LatencyTest-Topology来测试和解决这个问题。
我有一个喷嘴，它会发出随机值：
public void nextTuple() {
outputCollector.emit(new Values(Math.random()));
}
我有一个Bolt，它接收这些值，并在特定的时间睡眠。
public void execute(Tuple tuple) {
double input = tuple.getDoub
浏览 6提问于2017-08-18得票数 0
2回答
风暴HDFS螺栓不工作
、、、
所以我刚开始和风暴打交道，试着去理解它。我试图连接到kafka主题，读取数据并将其写入HDFS螺栓。一开始，我在没有shuffleGrouping("stormspout")的情况下创建了它，我的Storm显示喷口正在消耗来自主题的数据，但是没有任何东西被写入螺栓(除了它在HDFS上创建的空文件)。然后我添加了shuffleGrouping("stormspout")；现在螺栓似乎出现了错误。如果有人能帮忙的话，我会很感激的。
谢谢，科尔曼
误差
2015-04-13 00:02:58 s.k.PartitionManager INFO读取分区信息:/ s.k
浏览 1提问于2015-04-08得票数 0
我想使用kundera框架将一个对象插入到cassandra中，但是我收到了以下错误：
java.lang.RuntimeException: java.lang.ClassCastException: java.lang.String cannot be cast to [B
at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:448) ~[storm-core-1.0.0.jar:1.0.0]
at org.apache.storm.utils.D
浏览 1提问于2016-05-10得票数 0
我尝试了几个配置参数，甚至没有运气地使用withLocalModeOverride。我在这里错过了什么？
这里有一个示例应用程序，在30岁以后，计数器被重置，一切都重新启动。如果我能提供更多的细节，请告诉我。
用法：
mvn package
java -jar target/Test-0.1.0.jar
src/main/java/Test.java
package storm.test;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generate
浏览 8提问于2019-10-02得票数 0
回答已采纳
2回答
如何利用Twitter与风暴通量
、、、、
我正在尝试将一个项目从apache风暴迁移到twitter heron。经过很长时间的挣扎，我能够摆脱大部分的错误，比如使用className: "org.apache.storm.kafka.ZkHosts"而不是className: "storm.kafka.ZkHosts"。然而，我被困在提交拓扑。我使用Flux将拓扑提交给风暴。
当NullPointerException在ZkState中创建一个CuratorFramework对象时，我将得到它。在进一步的调查中，我发现了，它说，如果没有设置有关动物园管理员的配置，就会导致这个问题。
进一步调试，我发现
浏览 4提问于2017-07-17得票数 2
回答已采纳
我使用的导入org.apache.storm.kafka.bolt.KafkaProducerTopology的Maven依赖项是什么：
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
<version>1.0.1</version>
<scope>provided</scope>
</depen
浏览 11提问于2018-09-02得票数 0
回答已采纳
1回答
我试图将图像从base64解码为字节数组，并加载到S3桶中。贝娄是我的密码
byte[] imagecontent = Base64.decodeBase64(IMMAGE_CONTENT);
if(imagecontent .length>0) {
input = new ByteArrayInputStream(screenshot);
String file_name = "img_SOME_LOGIC.jpg";
conn.putObject( this.bucketname, file_name, input
浏览 0提问于2018-10-15得票数 0
回答已采纳
我们正在尝试在我们的Storm Topology中对性能进行基准测试。我们正在向Kafka主题摄取大约1000条/秒的消息。当我们将max.spout.pendind=2000放入KafkaSpout中时，我们在storm UI中看不到任何失败的消息，但当我们将max.spout.pendind值减少到500或100时，我们会在Storm UI的spout中看到许多失败的消息。我的理解是，如果我们保持低max.spout.pending，那么我们将不会有任何失败的消息，因为没有什么会超时，但它的行为是相反的。我们使用的是HDP 2.6.5版本的Storm 1.1.0版本。
我们有一个卡夫卡喷嘴
浏览 0提问于2018-09-15得票数 1
1回答
现在我想在一个污点中使用Drools，它在LocalCluster中正常工作，但是当我把它放到生产集群中时，它有错误。污点是：
public class DealLostBolt extends BaseRichBolt {
private static final long serialVersionUID = 1L;
private static final Logger LOGGER = LoggerFactory.getLogger("DEAL_LOST_BOLT");
private OutputCollector colle
浏览 6提问于2017-09-13得票数 0
回答已采纳
是否有用于Storm或Bolt的API来访问其ID？我所指的ID是传递给类String中的setSpout或setBolt方法的backtype.storm.topology.TopologyBuilder。
为日志记录目的访问此字符串将是有用的。
浏览 3提问于2014-12-18得票数 5
回答已采纳
1回答
我正在尝试将我的R脚本与Storm集成起来。我的Rbolt的代码是：
public class RBolt extends ShellBolt implements IRichBolt {
public RBolt() {
super("Rscript", "storm_OR.R");
}
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
outputFieldsDeclarer.declare(new Fields(&#
浏览 3提问于2016-03-05得票数 1
我使用Esper的开源版本来执行一些复杂的事件处理。我正在使用Esper + Storm的组合，这样我就可以在Storm bolt中编写我的CEP逻辑，它分布在网络上。
Esper还提供了高可用性的HA版本。我的问题是，使用Esper HA而不是Esper (开源)+ Storm组合有什么好处。
浏览 0提问于2015-03-13得票数 0
import backtype.storm.Config;`enter code here`
import backtype.storm.`StormSubmitter`;`enter code here
import backtype.storm.topology.TopologyBuilder;`enter code here`
import consumer.bolt.FirebaseBolt;`enter code here`
import consumer.bolt.WordCountBolt;
import consumer.bolt
浏览 0提问于2019-07-18得票数 0
我将多个元组发送到并行计数为8的bolt，然后处理，然后我想发出id，索引到下一个Bolt，它验证是否所有的元组都被处理了，并发送通知。怎么做呢？storm已经有它的实现了吗？
浏览 2提问于2017-05-27得票数 0
我已经设置了storm拓扑(1 worker)，其中spout(在java中)从redis出队(使用blpop)事件并传输到bolts。但一个观察是，当存在超过200万个队列，并且在风暴nimbus/supervisor/zookeeper/worker日志中没有发现警告/异常时，一些事件没有接收到闪存(在clojure中，6-spout线程，50-bolt线程)。
在本地，此方案不使用虚拟数据进行复制。在集群中没有看到网络延迟/丢包。平均处理延迟为100ms。如何在生产中找到原因来修复它。
(ns event-processor
(:import [backtype.storm Stor
浏览 10提问于2018-09-05得票数 1
回答已采纳
1回答
我有一个Storm拓扑，需要从Redis实例中流它的数据，我尝试运行从单个Redis实例读取的拓扑，但是似乎没有从Redis读取任何数据，而且当我检查返回的队列时，它是空的。我使用的是Storm版本0.9.3。
这是我的RedisQueueSpout，它是一个spout，它将使用指定的模式(又名键)将您的拓扑插入到Redis中，每次Storm轮询它时，它都会在该模式下查找输入数据。喷口发出一个带有ID信息的字段，发送到它后面的任何一个螺栓。
package storm.starter.spout;
import java.util.List;
import java.util.Map;
im
浏览 2提问于2016-05-04得票数 1
回答已采纳
我正在评估建立一个使用Storm (实时框架，而不是IDE)的网络爬虫。My Spout将提供一个URI流，这些URI必须在第一个Bolt中调用和检索，该Bolt是用php编写的。
我有node.js背景，我知道可以使用回调以非阻塞的方式构建这样的功能。我天真的想法是，只需要为Bolt配置很多任务，这样资源就可以在一些Tak等待答案的时候得到充分利用。问题是，php为每个进程分配大量资源是臭名昭著的，我不知道Storm是如何在内部管理的，这是不是一个好主意。
在这种环境中可能出现类似node.js的行为吗？或者我必须切换到另一种语言才能让这个Bolt实现它(如果是这样，我该怎么做)？
浏览 0提问于2012-03-29得票数 4
1回答
我想在一个operator.For示例中接收和处理三个流，在Storm中实现的代码如下所示：
builder.setBolt("C_bolt", C_bolt(), parallelism_hint) .fieldsGrouping("A_bolt", "TRAINING", new Fields("word")) .fieldsGrouping("B_bolt", "ANALYSIS", new Fields("word
浏览 0提问于2018-11-08得票数 0
回答已采纳
我是风暴的初学者。我正在尝试执行下面的示例程序
SampleSpout.java
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.s
浏览 2提问于2014-12-26得票数 0
回答已采纳
我刚刚学习了storm，并测试了来自的示例代码。它使用setDebug(true)来启用调试日志记录。
但我对以下调试信息感到困惑：
13867 [storm.starter.WordCountTopology.main()] INFO backtype.storm.daemon.nimbus -
Setting new assignment for topology id word-count-1-1403745226:
#backtype.storm.daemon.common.Assignment{:master-code-dir "/tmp/4ef1358b-92ce-40
浏览 0提问于2014-06-26得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
