Storm整合Kafka應用的開發
https://www.cnblogs.com/freeweb/p/5292961.html
Storm整合Kafka應用的開發
我們知道storm的作用主要是進行流式計算,對於源源不斷的均勻資料流流入處理是非常有效的,而現實生活中大部分場景並不是均勻的資料流,而是時而多時而少的資料流入,這種情況下顯然用批量處理是不合適的,如果使用storm做實時計算的話可能因為資料擁堵而導致伺服器掛掉,應對這種情況,使用kafka作為訊息佇列是非常合適的選擇,kafka可以將不均勻的資料轉換成均勻的訊息流,從而和storm比較完善的結合,這樣才可以實現穩定的流式計算,那麼我們接下來開發一個簡單的案例來實現storm和kafka的結合
storm和kafka結合,實質上無非是之前我們說過的計算模式結合起來,就是資料先進入kafka生產者,然後storm作為消費者進行消費,最後將消費後的資料輸出或者儲存到檔案、資料庫、分散式儲存等等,具體框圖如下:
這張圖片摘自部落格地址:http://www.cnblogs.com/tovin/p/3974417.html 在此感謝作者的奉獻
首先我們保證在伺服器上zookeeper、kafka、storm正常執行,然後我們開始寫程式,這裡使用eclipse for javaee IDE
和之前一樣,建立一個maven專案,在pom.xml寫入如下程式碼:
1 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 2 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 3 <modelVersion>4.0.0</modelVersion> 4 5 <groupId>kafkastorm</groupId> 6 <artifactId>kafkastorm</artifactId> 7 <version>0.0.1-SNAPSHOT</version> 8 <packaging>jar</packaging> 9 10 <name>kafkastorm</name> 11 <url>http://maven.apache.org</url> 12 13 <properties> 14 <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> 15 </properties> 16 17 <dependencies> 18 <dependency> 19 <groupId>junit</groupId> 20 <artifactId>junit</artifactId> 21 <version>3.8.1</version> 22 <scope>test</scope> 23 </dependency> 24 <dependency> 25 <groupId>org.apache.storm</groupId> 26 <artifactId>storm-core</artifactId> 27 <version>0.9.6</version> 28 <scope>provided</scope> 29 </dependency> 30 <dependency> 31 <groupId>org.apache.kafka</groupId> 32 <artifactId>kafka_2.9.2</artifactId> 33 <version>0.8.2.2</version> 34 <exclusions> 35 <exclusion> 36 <groupId>org.apache.zookeeper</groupId> 37 <artifactId>zookeeper</artifactId> 38 </exclusion> 39 <exclusion> 40 <groupId>log4j</groupId> 41 <artifactId>log4j</artifactId> 42 </exclusion> 43 </exclusions> 44 </dependency> 45 <dependency> 46 <groupId>org.apache.storm</groupId> 47 <artifactId>storm-kafka</artifactId> 48 <version>0.9.6</version> 49 </dependency> 50 </dependencies> 51 52 <build> 53 <plugins> 54 <plugin> 55 <artifactId>maven-assembly-plugin</artifactId> 56 <configuration> 57 <descriptorRefs> 58 <descriptorRef>jar-with-dependencies</descriptorRef> 59 </descriptorRefs> 60 </configuration> 61 <executions> 62 <execution> 63 <id>make-assembly</id> 64 <phase>package</phase> 65 <goals> 66 <goal>single</goal> 67 </goals> 68 </execution> 69 </executions> 70 </plugin> 71 </plugins> 72 </build> 73 </project>
主要是匯入的zookeeper、storm、kafka外部依賴這些疊加起來,還有<plugin>外掛便於我們後續對程式程序maven的打包
和之前一樣首先編寫storm消費kafka的邏輯,MessageScheme類,程式碼如下:
1 package net.zengzhiying; 2 3 import java.io.UnsupportedEncodingException; 4 import java.util.List; 5 6 import backtype.storm.spout.Scheme; 7 import backtype.storm.tuple.Fields; 8 import backtype.storm.tuple.Values; 9 10 public class MessageScheme implements Scheme { 11 12 public List<Object> deserialize(byte[] arg0) { 13 try { 14 String msg = new String(arg0, "UTF-8"); 15 return new Values(msg); 16 } catch (UnsupportedEncodingException e) { 17 e.printStackTrace(); 18 } 19 return null; 20 } 21 22 public Fields getOutputFields() { 23 return new Fields("msg"); 24 } 25 26 }
邏輯很簡單,就是對kafka出來的資料轉換成字串,接下來我們想辦法來處理strom清洗之後的資料,我們為了簡單就把輸出儲存到一個檔案中,Bolt邏輯SenqueceBolt類的程式碼如下:
1 package net.zengzhiying; 2 3 import java.io.DataOutputStream; 4 import java.io.FileNotFoundException; 5 import java.io.FileOutputStream; 6 import java.io.IOException; 7 8 import backtype.storm.topology.BasicOutputCollector; 9 import backtype.storm.topology.OutputFieldsDeclarer; 10 import backtype.storm.topology.base.BaseBasicBolt; 11 import backtype.storm.tuple.Fields; 12 import backtype.storm.tuple.Tuple; 13 import backtype.storm.tuple.Values; 14 15 public class SenqueceBolt extends BaseBasicBolt { 16 17 public void execute(Tuple arg0, BasicOutputCollector arg1) { 18 String word = (String) arg0.getValue(0); 19 String out = "output:" + word; 20 System.out.println(out); 21 22 //寫檔案 23 try { 24 DataOutputStream out_file = new DataOutputStream(new FileOutputStream("kafkastorm.out")); 25 out_file.writeUTF(out); 26 out_file.close(); 27 } catch (FileNotFoundException e) { 28 // TODO Auto-generated catch block 29 e.printStackTrace(); 30 } catch (IOException e) { 31 // TODO Auto-generated catch block 32 e.printStackTrace(); 33 } 34 35 arg1.emit(new Values(out)); 36 } 37 38 public void declareOutputFields(OutputFieldsDeclarer arg0) { 39 arg0.declare(new Fields("message")); 40 } 41 42 }
就是把輸出的訊息放到檔案kafkastorm.out中
然後我們編寫主類,也就是配置kafka提交topology到storm的程式碼,類名為StormKafkaTopo,程式碼如下:
1 package net.zengzhiying; 2 3 import java.util.HashMap; 4 import java.util.Map; 5 6 import backtype.storm.Config; 7 import backtype.storm.LocalCluster; 8 import backtype.storm.StormSubmitter; 9 import backtype.storm.generated.AlreadyAliveException; 10 import backtype.storm.generated.InvalidTopologyException; 11 import backtype.storm.spout.SchemeAsMultiScheme; 12 import backtype.storm.topology.TopologyBuilder; 13 import backtype.storm.utils.Utils; 14 import storm.kafka.BrokerHosts; 15 import storm.kafka.KafkaSpout; 16 import storm.kafka.SpoutConfig; 17 import storm.kafka.ZkHosts; 18 import storm.kafka.bolt.KafkaBolt; 19 20 public class StormKafkaTopo { 21 public static void main(String[] args) { 22 BrokerHosts brokerHosts = new ZkHosts("192.168.1.216:2181/kafka"); 23 24 SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, "topic1", "/kafka", "kafkaspout"); 25 26 Config conf = new Config(); 27 Map<String, String> map = new HashMap<String, String>(); 28 29 map.put("metadata.broker.list", "192.168.1.216:9092"); 30 map.put("serializer.class", "kafka.serializer.StringEncoder"); 31 conf.put("kafka.broker.properties", map); 32 conf.put("topic", "topic2"); 33 34 spoutConfig.scheme = new SchemeAsMultiScheme(new MessageScheme()); 35 36 TopologyBuilder builder = new TopologyBuilder(); 37 builder.setSpout("spout", new KafkaSpout(spoutConfig)); 38 builder.setBolt("bolt", new SenqueceBolt()).shuffleGrouping("spout"); 39 builder.setBolt("kafkabolt", new KafkaBolt<String, Integer>()).shuffleGrouping("bolt"); 40 41 if(args != null && args.length > 0) { 42 //提交到叢集執行 43 try { 44 StormSubmitter.submitTopology(args[0], conf, builder.createTopology()); 45 } catch (AlreadyAliveException e) { 46 e.printStackTrace(); 47 } catch (InvalidTopologyException e) { 48 e.printStackTrace(); 49 } 50 } else { 51 //本地模式執行 52 LocalCluster cluster = new LocalCluster(); 53 cluster.submitTopology("Topotest1121", conf, builder.createTopology()); 54 Utils.sleep(1000000); 55 cluster.killTopology("Topotest1121"); 56 cluster.shutdown(); 57 } 58 59 60 61 } 62 }
注意上面程式碼的配置,和之前單獨執行storm和kafka程式碼不太一樣,配置也很簡單,注意區別即可,如果細心的話會注意到這裡建了兩個topic一個是topic1,一個是topic2,topic1的含義kafka接收生產者過來的資料所需要的topic,topic2是KafkaBolt也就是storm中的bolt生成的topic,當然這裡topic2這行配置可以省略,是沒有任何問題的,類似於一箇中轉的東西,另外我們這次測試是上傳到伺服器執行,本地模式的程式碼沒有執行到,當然原理是一樣的
之前一般網上的教程到這裡就完畢了,這樣我們會引起一種沒有生產者的誤區,注意:上面3個類實現的功能是kafka消費者輸出的資料被storm消費!生產者的程式碼可以看成獨立的其他來源,可以寫在其他專案中,根據資料來源的情況來,下面我們為了示例,編寫一個類來進行生產,程式碼和之前kafka單獨的一樣:
1 package net.zengzhiying; 2 3 import java.util.ArrayList; 4 import java.util.List; 5 import java.util.Properties; 6 7 import kafka.javaapi.producer.Producer; 8 import kafka.producer.KeyedMessage; 9 import kafka.producer.ProducerConfig; 10 11 public class DataProducerInsert { 12 private static Producer<Integer,String> producer; 13 private final Properties props=new Properties(); 14 public DataProducerInsert(){ 15 //定義連線的broker list 16 props.put("metadata.broker.list", "192.168.1.216:9092"); 17 //定義序列化類 Java中物件傳輸之前要序列化 18 props.put("serializer.class", "kafka.serializer.StringEncoder"); 19 //props.put("advertised.host.name", "192.168.1.216"); 20 producer = new Producer<Integer, String>(new ProducerConfig(props)); 21 } 22 public static void main(String[] args) { 23 DataProducerInsert sp=new DataProducerInsert(); 24 //定義topic 25 String topic="topic1"; 26 //開始時間統計 27 long startTime = System.currentTimeMillis(); 28 //定義要傳送給topic的訊息 29 String messageStr = "This is a message"; 30 List<KeyedMessage<Integer, String>> datalist = new ArrayList<KeyedMessage<Integer, String>>(); 31 32 //構建訊息物件 33 KeyedMessage<Integer, String> data = new KeyedMessage<Integer, String>(topic, messageStr); 34 datalist.add(data); 35 36 //結束時間統計 37 long endTime = System.currentTimeMillis(); 38 KeyedMessage<Integer, String> data1 = new KeyedMessage<Integer, String>(topic, "用時" + (endTime-startTime)/1000.0); 39 datalist.add(data1); 40 41 //推送訊息到broker 42 producer.send(data); 43 producer.close(); 44 } 45 }
注意,這裡我們定義的topic是topic1,正好和前面的topic1資料來源對應,是整個kafka保持一致的topic,也就是說kafka生產者topic和消費者topic是必須名稱相同才可以響應,下面簡單添加了一點時間統計的程式碼,也很簡單
另外還要注意kafka配置檔案host.name儘量改成ip,和之前說過的一樣
到現在專案就編寫完成了,然後我們使用maven命令對專案打包,首先得保證我們windows上安裝好了maven,我們執行cmd,進入到當前專案目錄下,執行命令: mvn assembly:assembly 進行打包,打包的前提就是之前pom.xml的所有配置,執行後maven會自動下載相應的依賴並完成打包,需要耐心等待一會:
看到如圖所示的BUILD SUCCESS返回之後,那麼打包就成功了,現在進入專案目錄下的target目錄中,會看到2個jar包
其中後面那個檔名較長的大小也比較大,是包含相關依賴的包,接下來我們將這個包上傳到伺服器,然後使用storm執行jar包將我們的topology上傳到叢集中執行,注意是使用storm執行jar包,而不是java
/usr/storm/apache-storm-0.9.6/bin/storm jar kafkastorm-0.0.1-SNAPSHOT-jar-with-dependencies.jar net.zengzhiying.StormKafkaTopo kafkagostorm
前面是storm的絕對路徑,引數jar執行jar包,後面跟的是上傳topology的主類,最後kafkagostorm是我們上傳拓撲的名稱
這裡執行完之後會回到命令列,現在就在後臺叢集中開始分發運行了,這就是叢集模式,之前我們講的storm案例會不斷滾動大量資料,那個屬於本地模式,如果我們現在開啟ui介面的話,那麼訪問我們的地址http://192.168.1.216:8080/可以看到正在執行的Topology
可以看到狀態是active正在運行了,我們上面程式碼中kafkabolt建立了一個topic2的訊息,我們現在可以測試一下,消費者這裡只是簡單地原樣輸出,我們進入kafka目錄,執行下面命令:
bin/kafka-console-consumer.sh --zookeeper localhost:2181/kafka --topic topic2 --from-beginning
後面引數--from-beginning不新增也是可以的,新增包括舊資訊,不新增就是新的輸出
現在介面卡住,待會我們來觀察輸出,現在我們新開一個視窗,還是使用storm執行剛才的生產者類DataProducerInsert來生產一條訊息,命令如下:
/usr/storm/apache-storm-0.9.6/bin/storm jar kafkastorm-0.0.1-SNAPSHOT-jar-with-dependencies.jar net.zengzhiying.DataProducerInsert
回車之後,等待介面滾動2s程式跑完之後,我們看到另一個視窗輸出了訊息:
然後,我們的輸出檔案在哪呢,剛才我們使用storm執行的生產者程式碼,所以輸出的kafkastorm.out就在storm的安裝目錄下,我們使用cat或者vim都可以看到檔案內容,如果有時間統計的話兩行內容顯示可能會有點問題,因為後續要進行簡單的轉換,去掉時間統計程式碼只輸出訊息的內容如下,這裡使用vim開啟的:
另外注意,上傳拓撲時所有的程式碼都載入到叢集了,所以修改程式碼版本時,一定要先在storm目錄下執行 bin/storm kill topo_name 結束拓撲,修改程式碼後重新上傳即可再次執行,否則可能會出現錯誤,在叢集上的時候kafka配置檔案的host.name註釋即可,預設為localhost,最後程式碼中用到的引數比較多,很容易出錯,所以寫程式碼時還是要仔細點
這樣storm整合kafka的測試案例就完成了,並且實現了一定的功能,只要我們靈活掌握了怎麼寫kafka和storm結合的整體拓撲結構,那麼主要的程式碼就集中在資料來源也就是kafka生產者的傳送和storm消費後的儲存問題,這所有的程式碼都是在storm和kafka給好的方法內寫邏輯,而不用關心底層,這樣使開發更加簡單快捷,比如我們消費後的資料既可以寫到檔案、資料庫還可以索引到solr,存到Hbase等,這樣就可以靈活運用了;其實最關鍵的還是要了解這些框架底層的實現原理,這樣遇到問題才可以知其然知其所以然