使用Storm Toplology從卡夫卡隊列中消費數據

算法使用KafkaSpout從卡夫卡隊列中讀取數據。使用Storm Toplology從卡夫卡隊列中消費數據

我面臨以下異常：

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和Storm JVM（本地執行）。

來源

2015-12-17 SriniReddy M

我無法重現錯誤。順便說一句：你的代碼堆棧跟蹤不匹配：在你的代碼中，你使用'LocalCluster'，但你的堆棧跟蹤說'StormSubmitter'。在這種情況下，它確實失敗了？你是否嘗試從一個空的Config開始，逐個添加參數以查看哪一個產生問題？ –

@Sax我沒有這樣試過，我現在就試試。我只是在亞馬遜aws集羣中運行這個類文件。 –

試試這個：

List<String> hosts = Arrays.asList("localhost") 
conf.put(Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, hosts);

我覺得是從Arrays.asList("localhost")返回java.util.AbstractList<Object>問題，解析爲JSON。

來源

2015-12-17 03:42:33 salc2

沒有面臨相同的錯誤。 –

使用Storm Toplology從卡夫卡隊列中消費數據

回答

相關問題