我正试图重新平衡我的风暴拓扑结构,这是使用Kafka。我的代码是:
TopologyBuilder builder = new TopologyBuilder();
Properties kafkaProps = new Properties();
kafkaProps.put("zk.connect", "localhost:2181");
kafkaProps.put("zk.connectiontimeout.ms", "1000000");
kafkaProps.put("groupid", "storm");
builder.setSpout( "kafkaSpout" , new KafkaSpout(kafkaProps, "test"), 3);
builder.setBolt( "eventBolt", new EventBolt(), 2 ).shuffleGrouping( "kafkaSpout", "eventStream" );
builder.setBolt( "tableBolt", new TableBolt(), 2 ).shuffleGrouping( "kafkaSpout", "tableStream");
Map<String, Object> conf = new HashMap<String, Object>();
conf.put(Config.TOPOLOGY_DEBUG, true);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("test", conf, builder.createTopology());
Utils.sleep( 1000*5 );
List<TopologySummary> topologySummaries = cluster.getClusterInfo().get_topologies();
for ( TopologySummary summary : topologySummaries ) {
StormTopology topology = cluster.getTopology( summary.get_id() );
RebalanceOptions options = new RebalanceOptions();
options.set_wait_secs( 0 );
options.set_num_workers( 4 );
for ( String name : topology.get_bolts().keySet() ) {
System.err.println( name + " " + topology.get_bolts().get(name).get_common().get_json_conf() );
options.put_to_num_executors( name , 5);
}
for ( String name : topology.get_spouts().keySet() ) {
System.err.println( name + " " + topology.get_spouts().get(name).get_common().get_json_conf() );
options.put_to_num_executors( name , 5);
}
cluster.rebalance( summary.get_name() , options);
}
但是,在重新平衡过程中,会显示以下错误跟踪:
10341 [storm_rishabh-1361473654345-95461d10_watcher_executor] INFO kafka.consumer.ZookeeperConsumerConnector - storm_rishabh-1361473654345-95461d10 begin rebalancing consumer storm_rishabh-1361473654345-95461d10 try #1
10341 [storm_rishabh-1361473654345-3b26ed76_watcher_executor] INFO kafka.consumer.ZookeeperConsumerConnector - storm_rishabh-1361473654345-3b26ed76 begin rebalancing consumer storm_rishabh-1361473654345-3b26ed76 try #1
10342 [storm_rishabh-1361473654345-95461d10_watcher_executor] ERROR kafka.consumer.ZookeeperConsumerConnector - storm_rishabh-1361473654345-95461d10 error during syncedRebalance
java.lang.NullPointerException: null
at kafka.utils.ZkUtils$.getChildrenParentMayNotExist(ZkUtils.scala:181) ~[kafka_2.9.2-0.7.0.jar:na]
at kafka.utils.ZkUtils$.getCluster(ZkUtils.scala:202) ~[kafka_2.9.2-0.7.0.jar:na]
at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply$mcVI$sp(ZookeeperConsumerConnector.scala:447) ~[kafka_2.9.2-0.7.0.jar:na]
at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:78) ~[scala-library-2.9.2.jar:na]
at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:444) ~[kafka_2.9.2-0.7.0.jar:na]
at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anon$1.run(ZookeeperConsumerConnector.scala:401) ~[kafka_2.9.2-0.7.0.jar:na]
10342 [storm_rishabh-1361473654345-3b26ed76_watcher_executor] ERROR kafka.consumer.ZookeeperConsumerConnector - storm_rishabh-1361473654345-3b26ed76 error during syncedRebalance
java.lang.NullPointerException: null
at kafka.utils.ZkUtils$.getChildrenParentMayNotExist(ZkUtils.scala:181) ~[kafka_2.9.2-0.7.0.jar:na]
at kafka.utils.ZkUtils$.getCluster(ZkUtils.scala:202) ~[kafka_2.9.2-0.7.0.jar:na]
at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply$mcVI$sp(ZookeeperConsumerConnector.scala:447) ~[kafka_2.9.2-0.7.0.jar:na]
at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:78) ~[scala-library-2.9.2.jar:na]
at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:444) ~[kafka_2.9.2-0.7.0.jar:na]
at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anon$1.run(ZookeeperConsumerConnector.scala:401) ~[kafka_2.9.2-0.7.0.jar:na]
10342 [storm_rishabh-1361473654345-95461d10_watcher_executor] INFO kafka.consumer.ZookeeperConsumerConnector - storm_rishabh-1361473654345-95461d10 stopping watcher executor thread for consumer storm_rishabh-1361473654345-95461d10
10343 [storm_rishabh-1361473654345-3b26ed76_watcher_executor] INFO kafka.consumer.ZookeeperConsumerConnector - storm_rishabh-1361473654345-3b26ed76 stopping watcher executor thread for consumer storm_rishabh-1361473654345-3b26ed76
有人能告诉我有什么问题吗?我是否需要在kafkaspout中定义更多的东西,以便在重新平衡时,它正确地关闭,然后重新开始?
2条答案
按热度按时间vohkndzv1#
我在跑步时也遇到了同样的问题
LocalCluster
(用于开发目的)。我更改了我的测试配置yaml,将worker的数量减少到1:这就解决了问题。我还没有尝试在一个实际的分布式集群上运行它,所以我不知道这是否只是一个运行的工件
LocalCluster
模式与否。(在我的代码中,我从不调用
LocalCluster.rebalance
.)eeq64g8w2#
使用来自主管或nimbus节点的storm rebalance命令。
例如,storm rebalance mytopology-n5-e blue spout=3-e yellow bolt=10。
请参考本网站。www.michael-noll.com.