Storm(七)Storm整合kafka
阿新 • • 發佈:2018-12-11
使用kafka-client jar進行Storm Apache Kafka整合
這包括新的Apache Kafka消費者API。相容性 Apache Kafka版本0.10起 引入jar包
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka-client</artifactId>
<version>1.2.0</version>
</dependency>
從kafka中訂閱訊息讀取
通過使用KafkaSpoutConfig類來配置spout實現。此類使用Builder模式,可以通過呼叫其中一個Builders建構函式或通過呼叫KafkaSpoutConfig類中的靜態方法構建器來啟動。
用法示例
建立一個簡單的不kafka資料來源 以下將使用釋出到“topic”的所有事件,並將它們傳送到MyBolt,其中包含“topic”,“partition”,“offset”,“key”,“value”欄位。
TopologyBuilder tp = new TopologyBuilder(); tp.setSpout("kafka_spout", new KafkaSpout(KafkaSpoutConfig.builder("localhost:9092" , "qxw").build()), 1); tp.setBolt("bolt", new MyBolt()).shuffleGrouping("kafka_spout"); Config cfg=new Config(); cfg.setNumWorkers(1);//指定工作程序數 (jvm數量,分散式環境下可用,本地模式設定無意義) cfg.setDebug(true); LocalCluster locl=new LocalCluster(); locl.submitTopology("kkafka-topo",cfg,tp.createTopology());
public static class MyBolt extends BaseBasicBolt{ public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) { System.err.println("接受訂閱kafka訊息: "+tuple.getStringByField("topic")); System.err.println("接受訂閱kafka訊息: "+tuple.getStringByField("value")); } public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { } }