Storm入門 第二章准備開始

杜老師說 2022-01-07 13:10:34 阅读数:606

storm 第二章 第二

本文翻譯自《Getting Started With Storm》  譯者:吳京潤   編輯:方騰飛

准備開始

在本章,我們要創建一個Storm工程和我們的第一個Storm拓撲結構。

NOTE: 下面假設你的JRE版本在1.6以上。我們推薦Oracle提供的JRE。你可以到http://www.java .com/downloads/下載。

操作模式

開始之前,有必要了解一下Storm的操作模式。有下面兩種方式。

本地模式

在本地模式下,Storm拓撲結構運行在本地計算機的單一JVM進程上。這個模式用於開發、測試以及調試,因為這是觀察所有組件如何協同工作的最簡單方法。在這種模式下,我們可以調整參數,觀察我們的拓撲結構如何在不同的Storm配置環境下運行。要在本地模式下運行,我們要下載Storm開發依賴,以便用來開發並測試我們的拓撲結構。我們創建了第一個Storm工程以後,很快就會明白如何使用本地模式了。

NOTE: 在本地模式下,跟在集群環境運行很像。不過很有必要確認一下所有組件都是線程安全的,因為當把它們部署到遠程模式時它們可能會運行在不同的JVM進程甚至不同的物理機上,這個時候它們之間沒有直接的通訊或共享內存。

我們要在本地模式運行本章的所有例子。

遠程模式

在遠程模式下,我們向Storm集群提交拓撲,它通常由許多運行在不同機器上的流程組成。遠程模式不會出現調試信息, 因此它也稱作生產模式。不過在單一開發機上建立一個Storm集群是一個好主意,可以在部署到生產環境之前,用來確認拓撲在集群環境下沒有任何問題。

你將在第六章學到更多關於遠程模式的內容,並在附錄B學到如何安裝一個Storm集群。

Hello World

我們在這個工程裏創建一個簡單的拓撲,數單詞數量。我們可以把這個看作Storm的“Hello World”。不過,這是一個非常强大的拓撲,因為它能够擴展到幾乎無限大的規模,而且只需要做一些小修改,就能用它構建一個統計系統。舉個例子,我們可以修改一下工程用來找出Twitter上的熱點話題。

要創建這個拓撲,我們要用一個spout讀取文本,第一個bolt用來標准化單詞,第二個bolt為單詞計數,如圖2-1所示。

你可以從這個網址下載源碼壓縮包, https://github.com/storm-book/examples-ch02-getting_started/zipball/master

NOTE: 如果你使用git(一個分布式版本控制與源碼管理工具),你可以執行git clone [email protected]:storm-book/examples-ch02-getting_started.git,把源碼檢出到你指定的目錄。

Java安裝檢查

構建Storm運行環境的第一步是檢查你安裝的Java版本。打開一個控制臺窗口並執行命令:java -version。控制臺應該會顯示出類似如下的內容:

 java -version java version "1.6.0_26" Java(TM) SE Runtime Enviroment (build 1.6.0_26-b03) Java HotSpot(TM) Server VM (build 20.1-b02, mixed mode)

如果不是上述內容,檢查你的Java安裝情况。(參考http://www.java.com/download/

創建工程

開始之前,先為這個應用建一個目錄(就像你平常為Java應用做的那樣)。這個目錄用來存放工程源碼。

接下來我們要下載Storm依賴包,這是一些jar包,我們要把它們添加到應用類路徑中。你可以采用如下兩種方式之一完成這一步:

  • 下載所有依賴,解壓縮它們,把它 們添加到類路徑
  • 使用Apache Maven

NOTE: Maven是一個軟件項目管理的綜合工具。它可以用來管理項目的開發周期的許多方面,從包依賴到版本發布過程。在這本書中,我們將廣泛使用它。如果要檢查是否已經安裝了maven,在命令行運行mvn。如果沒有安裝你可以從http://maven.apache.org/download.html下載。

沒有必要先成為一個Maven專家才能使用Storm,不過了解一下關於Maven工作方式的基礎知識仍然會對你有所幫助。你可以在Apache Maven的網站上找到更多的信息(http://maven.apache.org/)。

NOTE: Storm的Maven依賴引用了運行Storm本地模式的所有庫。

要運行我們的拓撲,我們可以編寫一個包含基本組件的pom.xml文件。

 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>storm.book</groupId> <artifactId>Getting-Started</artifactId> <version>0.0.1-SNAPSHOT</version> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>2.3.2</version> <configuration> <source>1.6</source> <target>1.6</target> <compilerVersion>1.6</compilerVersion> </configuration> </plugin> </plugins> </build> <repositories> <!-- Repository where we can found the storm dependencies --> <repository> <id>clojars.org</id> <url>http://clojars.org/repo</url> </repository> </repositories> <dependencies> <!-- Storm Dependency --> <dependency> <groupId>storm</groupId> <artifactId>storm</artifactId> <version>0.6.0</version> </dependency> </dependencies> </project>

開頭幾行指定了工程名稱和版本號。然後我們添加了一個編譯器插件,告知Maven我們的代碼要用Java1.6編譯。接下來我們定義了Maven倉庫(Maven支持為同一個工程指定多個倉庫)。clojars是存放Storm依賴的倉庫。Maven會為運行本地模式自動下載必要的所有子包依賴。

一個典型的Maven Java工程會擁有如下結構:

 我們的應用目錄/ ├── pom.xml └── src └── main └── java | ├── spouts | └── bolts └── resources

java目錄下的子目錄包含我們的代碼,我們把要統計單詞數的文件保存在resource目錄下。

NOTE:命令mkdir -p 會創建所有需要的父目錄。

創建我們的第一個Topology

我們將為運行單詞計數創建所有必要的類。可能這個例子中的某些部分,現在無法講的很清楚,不過我們會在隨後的章節做進一步的講解。

Spout

pout WordReader類實現了IRichSpout接口。我們將在第四章看到更多細節。WordReader負責從文件按行讀取文本,並把文本行提供給第一個bolt

NOTE: 一個spout發布一個定義域列錶。這個架構允許你使用不同的bolts從同一個spout流讀取數據,它們的輸出也可作為其它bolts的定義域,以此類推。

例2-1包含WordRead類的完整代碼(我們將會分析下述代碼的每一部分)。

 /** * 例2-1.src/main/java/spouts/WordReader.java */ package spouts; import java.io.BufferedReader; import java.io.FileNotFoundException; import java.io.FileReader; import java.util.Map; import backtype.storm.spout.SpoutOutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.IRichSpout; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Values; public class WordReader implements IRichSpout { private SpoutOutputCollector collector; private FileReader fileReader; private boolean completed = false; private TopologyContext context; public boolean isDistributed() {return false;} public void ack(Object msgId) { System.out.println("OK:"+msgId); } public void close() {} public void fail(Object msgId) { System.out.println("FAIL:"+msgId); } /** * 這個方法做的惟一一件事情就是分發文件中的文本行 */ public void nextTuple() { /** * 這個方法會不斷的被調用,直到整個文件都讀完了,我們將等待並返回。 */ if(completed){ try { Thread.sleep(1000); } catch (InterruptedException e) { //什麼也不做 } return; } String str; //創建reader BufferedReader reader = new BufferedReader(fileReader); try{ //讀所有文本行 while((str = reader.readLine()) != null){ /** * 按行發布一個新值 */ this.collector.emit(new Values(str),str); } }catch(Exception e){ throw new RuntimeException("Error reading tuple",e); }finally{ completed = true; } } /** * 我們將創建一個文件並維持一個collector對象 */ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { try { this.context = context; this.fileReader = new FileReader(conf.get("wordsFile").toString()); } catch (FileNotFoundException e) { throw new RuntimeException("Error reading file ["+conf.get("wordFile")+"]"); } this.collector = collector; } /** * 聲明輸入域"word" */ public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("line")); } }

第一個被調用的spout方法都是public void open(Map conf, TopologyContext context, SpoutOutputCollector collector)。它接收如下參數:配置對象,在定義topology對象是創建;TopologyContext對象,包含所有拓撲數據;還有SpoutOutputCollector對象,它能讓我們發布交給bolts處理的數據。下面的代碼主是這個方法的實現。

 public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { try { this.context = context; this.fileReader = new FileReader(conf.get("wordsFile").toString()); } catch (FileNotFoundException e) { throw new RuntimeException("Error reading file ["+conf.get("wordFile")+"]"); } this.collector = collector; }

我們在這個方法裏創建了一個FileReader對象,用來讀取文件。接下來我們要實現public void nextTuple(),我們要通過它向bolts發布待處理的數據。在這個例子裏,這個方法要讀取文件並逐行發布數據。

 public void nextTuple() { if(completed){ try { Thread.sleep(1); } catch (InterruptedException e) { //什麼也不做 } return; } String str; BufferedReader reader = new BufferedReader(fileReader); try{ while((str = reader.readLine()) != null){ this.collector.emit(new Values(str)); } }catch(Exception e){ throw new RuntimeException("Error reading tuple",e); }finally{ completed = true; } }

NOTE: Values是一個ArrarList實現,它的元素就是傳入構造器的參數。

nextTuple()會在同一個循環內被ack()fail()周期性的調用。沒有任務時它必須釋放對線程的控制,其它方法才有機會得以執行。因此nextTuple的第一行就要檢查是否已處理完成。如果完成了,為了降低處理器負載,會在返回前休眠一毫秒。如果任務完成了,文件中的每一行都已被讀出並分發了。

NOTE:元組(tuple)是一個具名值列錶,它可以是任意java對象(只要它是可序列化的)。默認情况,Storm會序列化字符串、字節數組、ArrayList、HashMap和HashSet等類型。

Bolts

現在我們有了一個spout,用來按行讀取文件並每行發布一個元組,還要創建兩個bolts,用來處理它們(看圖2-1)。bolts實現了接口backtype.storm.topology.IRichBolt

bolt最重要的方法是void execute(Tuple input),每次接收到元組時都會被調用一次,還會再發布若幹個元組。

NOTE: 只要必要,boltspout會發布若幹元組。當調用nextTupleexecute方法時,它們可能會發布0個、1個或許多個元組。你將在第五章學習更多這方面的內容。

第一個boltWordNormalizer,負責得到並標准化每行文本。它把文本行切分成單詞,大寫轉化成小寫,去掉頭尾空白符。

首先我們要聲明bolt的出參:

 public void declareOutputFields(OutputFieldsDeclarer declarer){ declarer.declare(new Fields("word")); }

這裏我們聲明bolt將發布一個名為“word”的域。

下一步我們實現public void execute(Tuple input),處理傳入的元組:

 public void execute(Tuple input){ String sentence=input.getString(0); String[] words=sentence.split(" "); for(String word : words){ word=word.trim(); if(!word.isEmpty()){ word=word.toLowerCase(); //發布這個單詞 collector.emit(new Values(word)); } } //對元組做出應答 collector.ack(input); }

第一行從元組讀取值。值可以按比特置或名稱讀取。接下來值被處理並用collector對象發布。最後,每次都調用collector對象的ack()方法確認已成功處理了一個元組。

例2-2是這個類的完整代碼。

 //例2-2 src/main/java/bolts/WordNormalizer.java package bolts; import java.util.ArrayList; import java.util.List; import java.util.Map; import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.IRichBolt; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; import backtype.storm.tuple.Values; public class WordNormalizer implements IRichBolt{ private OutputCollector collector; public void cleanup(){} /** * *bolt*從單詞文件接收到文本行,並標准化它。 * 文本行會全部轉化成小寫,並切分它,從中得到所有單詞。 */ public void execute(Tuple input){ String sentence = input.getString(0); String[] words = sentence.split(" "); for(String word : words){ word = word.trim(); if(!word.isEmpty()){ word=word.toLowerCase(); //發布這個單詞 List a = new ArrayList(); a.add(input); collector.emit(a,new Values(word)); } } //對元組做出應答 collector.ack(input); } public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { this.collector=collector; } /** * 這個*bolt*只會發布“word”域 */ public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word")); } }

NOTE:通過這個例子,我們了解了在一次execute調用中發布多個元組。如果這個方法在一次調用中接收到句子“This is the Storm book”,它將會發布五個元組。

下一個boltWordCounter,負責為單詞計數。這個拓撲結束時(cleanup()方法被調用時),我們將顯示每個單詞的數量。

NOTE: 這個例子的bolt什麼也沒發布,它把數據保存在map裏,但是在真實的場景中可以把數據保存到數據庫。

package bolts;import java.util.HashMap;import java.util.Map;import backtype.storm.task.OutputCollector;import backtype.storm.task.TopologyContext;import backtype.storm.topology.IRichBolt;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.tuple.Tuple;public class WordCounter implements IRichBolt{ Integer id; String name; Map<String,Integer> counters; private OutputCollector collector; /** * 這個spout結束時(集群關閉的時候),我們會顯示單詞數量 */ @Override public void cleanup(){ System.out.println("-- 單詞數 【"+name+"-"+id+"】 --"); for(Map.Entry<String,Integer> entry : counters.entrySet()){ System.out.println(entry.getKey()+": "+entry.getValue()); } } /** * 為每個單詞計數 */ @Override public void execute(Tuple input) { String str=input.getString(0); /** * 如果單詞尚不存在於map,我們就創建一個,如果已在,我們就為它加1 */ if(!counters.containsKey(str)){ conters.put(str,1); }else{ Integer c = counters.get(str) + 1; counters.put(str,c); } //對元組作為應答 collector.ack(input); } /** * 初始化 */ @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector){ this.counters = new HashMap<String, Integer>(); this.collector = collector; this.name = context.getThisComponentId(); this.id = context.getThisTaskId(); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) {}}

execute方法使用一個map收集單詞並計數。拓撲結束時,將調用clearup()方法打印計數器map。(雖然這只是一個例子,但是通常情况下,當拓撲關閉時,你應當使用cleanup()方法關閉活動的連接和其它資源。)

主類

你可以在主類中創建拓撲和一個本地集群對象,以便於在本地測試和調試。LocalCluster可以通過Config對象,讓你嘗試不同的集群配置。比如,當使用不同數量的工作進程測試你的拓撲時,如果不小心使用了某個全局變量或類變量,你就能够發現錯誤。(更多內容請見第三章

NOTE:所有拓撲節點的各個進程必須能够獨立運行,而不依賴共享數據(也就是沒有全局變量或類變量),因為當拓撲運行在真實的集群環境時,這些進程可能會運行在不同的機器上。

接下來,TopologyBuilder將用來創建拓撲,它决定Storm如何安排各節點,以及它們交換數據的方式。

 TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("word-reader", new WordReader()); builder.setBolt("word-normalizer", new WordNormalizer()).shuffleGrouping("word-reader"); builder.setBolt("word-counter", new WordCounter()).shuffleGrouping("word-normalizer");

spoutbolts之間通過shuffleGrouping方法連接。這種分組方式决定了Storm會以隨機分配方式從源節點向目標節點發送消息。

下一步,創建一個包含拓撲配置的Config對象,它會在運行時與集群配置合並,並通過prepare方法發送給所有節點。

 Config conf = new Config(); conf.put("wordsFile", args[0]); conf.setDebug(true);

spout讀取的文件的文件名,賦值給wordFile屬性。由於是在開發階段,設置debug屬性為true,Strom會打印節點間交換的所有消息,以及其它有助於理解拓撲運行方式的調試數據。

正如之前講過的,你要用一個LocalCluster對象運行這個拓撲。在生產環境中,拓撲會持續運行,不過對於這個例子而言,你只要運行它幾秒鐘就能看到結果。

 LocalCluster cluster = new LocalCluster(); cluster.submitTopology("Getting-Started-Topologie", conf, builder.createTopology()); Thread.sleep(2000); cluster.shutdown();

調用createTopologysubmitTopology,運行拓撲,休眠兩秒鐘(拓撲在另外的線程運行),然後關閉集群。

例2-3是完整的代碼

 //例2-3 src/main/java/TopologyMain.java import spouts.WordReader; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.topology.TopologyBuilder; import backtype.storm.tuple.Fields; import bolts.WordCounter; import bolts.WordNormalizer; public class TopologyMain { public static void main(String[] args) throws InterruptedException { //定義拓撲 TopologyBuilder builder = new TopologyBuilder()); builder.setSpout("word-reader", new WordReader()); builder.setBolt("word-normalizer", new WordNormalizer()).shuffleGrouping("word-reader"); builder.setBolt("word-counter", new WordCounter(),2).fieldsGrouping("word-normalizer", new Fields("word")); //配置 Config conf = new Config(); conf.put("wordsFile", args[0]); conf.setDebug(false); //運行拓撲 conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("Getting-Started-Topologie", conf, builder.createTopology(); Thread.sleep(1000); cluster.shutdown(); } }

觀察運行情况

你已經為運行你的第一個拓撲准備好了。在這個目錄下面創建一個文件,/src/main/resources/words.txt,一個單詞一行,然後用下面的命令運行這個拓撲:mvn exec:java -Dexec.mainClass=”TopologyMain” -Dexec.args=”src/main/resources/words.txt

舉個例子,如果你的words.txt文件有如下內容: Storm test are great is an Storm simple application but very powerful really Storm is great 你應該會在日志中看到類似下面的內容: is: 2 application: 1 but: 1 great: 1 test: 1 simple: 1 Storm: 3 really: 1 are: 1 great: 1 an: 1 powerful: 1 very: 1 在這個例子中,每類節點只有一個實例。但是如果你有一個非常大的日志文件呢?你能够很輕松的改變系統中的節點數量實現並行工作。這個時候,你就要創建兩個WordCounter實例。

 builder.setBolt("word-counter", new WordCounter(),2).shuffleGrouping("word-normalizer");

程序返回時,你將看到: — 單詞數 【word-counter-2】 — application: 1 is: 1 great: 1 are: 1 powerful: 1 Storm: 3 — 單詞數 [word-counter-3] — really: 1 is: 1 but: 1 great: 1 test: 1 simple: 1 an: 1 very: 1 棒極了!修改並行度實在是太容易了(當然對於實際情况來說,每個實例都會運行在單獨的機器上)。不過似乎有一個問題:單詞isgreat分別在每個WordCounter各計數一次。怎麼會這樣?當你調用shuffleGrouping時,就决定了Storm會以隨機分配的方式向你的bolt實例發送消息。在這個例子中,理想的做法是相同的單詞問題發送給同一個WordCounter實例。你把shuffleGrouping(“word-normalizer”)換成fieldsGrouping(“word-normalizer”, new Fields(“word”))就能達到目的。試一試,重新運行程序,確認結果。 你將在後續章節學習更多分組方式和消息流類型。

結論 

我們已經討論了Storm的本地和遠程操作模式之間的不同,以及Storm的强大和易於開發的特性。你也學習了一些Storm的基本概念,我們將在後續章節深入講解它們。

 

原創文章,轉載請注明: 轉載自並發編程網 – ifeve.com本文鏈接地址: Storm入門 第二章准備開始

FavoriteLoading添加本文到我的收藏
版权声明:本文为[杜老師說]所创,转载请带上原文链接,感谢。 https://gsmany.com/2022/01/202201071310334418.html