该算法使用 KafkaSpout
.
我面临以下例外情况:
Exception in thread "main" java.lang.IllegalStateException: Couldn't initialize the topology
at com.bridgera.iot.kafka.App.main(App.java:63)
Caused by: java.lang.IllegalArgumentException: Storm conf is not valid. Must be json-serializable
at backtype.storm.StormSubmitter.submitTopology(StormSubmitter.java:104)
at backtype.storm.StormSubmitter.submitTopology(StormSubmitter.java:70)
at com.bridgera.iot.kafka.App.main(App.java:60)
我的java代码:
public static void main( String[] args ) throws AlreadyAliveException, InvalidTopologyException
{
String nimbusHost = "localhost";
ZkHosts zkHosts=new ZkHosts("localhost:2181");
String topic_name="test";
String consumer_group_id="storm";
String zookeeper_root="";
SpoutConfig kafkaConfig=new SpoutConfig(zkHosts,
topic_name, zookeeper_root, consumer_group_id);
kafkaConfig.scheme=new SchemeAsMultiScheme(new StringScheme());
KafkaSpout kafkaSpout = new KafkaSpout(kafkaConfig);
TopologyBuilder builder=new TopologyBuilder();
//builder.setSpout("KafkaSpout", kafkaSpout, 1);
builder.setSpout("KafkaSpout", kafkaSpout);
builder.setBolt("PrinterBolt", new PrinterBolt()).globalGrouping("KafkaSpout");
Map<String, Object> conf = new HashMap<String, Object>();
conf.put(Config.TRANSACTIONAL_ZOOKEEPER_PORT, 2181);
conf.put(Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, Arrays.asList("localhost"));
conf.put(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT, 20000);
conf.put(Config.STORM_ZOOKEEPER_CONNECTION_TIMEOUT, 20000);
conf.put(Config.STORM_ZOOKEEPER_RETRY_TIMES, 3);
conf.put(Config.STORM_ZOOKEEPER_RETRY_INTERVAL, 30);
LocalCluster cluster=new LocalCluster();
try{
cluster.submitTopology("KafkaConsumerTopology", conf, builder.createTopology());
Thread.sleep(120000);
}catch (Exception e) {
throw new IllegalStateException("Couldn't initialize the topology", e);
}
}
让我知道在配置中我做错了什么。仅供参考:我正在aws集群中运行zookeeper和stormjvm(本地执行)。
1条答案
按热度按时间zf9nrax11#
试试这个:
我认为这是个问题
java.util.AbstractList<Object>
从返回Arrays.asList("localhost")
以及对json的解析。