這篇文章將為大家詳細講解有關(guān)windows如何安裝storm eclipse調(diào)試TopN實例,小編覺得挺實用的,因此分享給大家做個參考,希望大家閱讀完這篇文章后可以有所收獲。
成都創(chuàng)新互聯(lián)公司專注于通道網(wǎng)站建設服務及定制,我們擁有豐富的企業(yè)做網(wǎng)站經(jīng)驗。 熱誠為您提供通道營銷型網(wǎng)站建設,通道網(wǎng)站制作、通道網(wǎng)頁設計、通道網(wǎng)站官網(wǎng)定制、微信平臺小程序開發(fā)服務,打造通道網(wǎng)絡公司原創(chuàng)品牌,更為您提供通道網(wǎng)站排名全網(wǎng)營銷落地服務。
配置Java環(huán)境變量 JAVA_HOME、Path、CLASSPATH三個值分別為(按照自己安裝狀況設置,此處供參考):
D:\java\jdk1.8
%JAVA_HOME%/bin;%JAVA_HOME%/jre/bin
.;%JAVA_HOME%/lib/dt.jar;%JAVA_HOME%/lib/tools.jar (要加.表示當前路徑)
這是為了測試安裝效果,我們將部署 storm-starter project案例中word coun程序,用的是python寫的multi-lang bolt,使用python 2.7.11,安裝路徑在:
C:\Python27\
Download Apache Zookeeper 3.4.8 ,解壓配置:
> cd zookeeper-3.4.8
> copy conf\zoo_sample.cfg conf\zoo.cfg
> .\bin\zkServer.cmd
Storm的windows官方版還沒有釋放,here.下載,源碼here下載。
注意1:
源碼一定要用這個版本,否則啟動會報各種錯誤,而這些錯誤和 jdk、python、zookeeper、eclipse 版本都無關(guān)。
http://dl.dropboxusercontent.com/s/iglqz73chkul1tu/storm-0.9.1-incubating-SNAPSHOT-12182013.zip
配置Storm環(huán)境變量
Storm需要STORM_HOME和JAVA_HOME,比如STORM_HOME為:
C:\storm-0.9.1-incubating-SNAPSHOT-12182013\
在PATH中加入:
%STORM_HOME%\bin;C:\Python27\Lib\site-packages\;C:\Python27\Scripts\
此處與參考文章略有不同,下圖是參考文章給出的配置
JAVA_HOME已經(jīng)在安裝JDK時手動配置了環(huán)境變量,而Python好像是默認自動就會配置好環(huán)境變量的,
我的Python目錄下沒有Scripts文件夾,暫時這樣配置就可以了,不影響下面的使用。
Nimbus
注意2:
一定要在 STORM_HOME 目錄下執(zhí)行后續(xù)命令,否則會報錯。
ERROR backtype.storm.event - Error when processing event
java.lang.RuntimeException: java.io.InvalidClassException: clojure.lang.APersistentMap; local class incompatible: stream classdesc serialVersionUID = 8648225932767613808, local class serialVersionUID = 270281984708184947
at backtype.storm.utils.Utils.deserialize(Utils.java:86) ~[storm-core-0.9.1-incubating-SNAPSHOT-12182013.jar:na]
> cd %STORM_HOME%
> storm nimbus
Supervisor
> cd %STORM_HOME%
> storm supervisor
Storm UI # 可選,也可以用 storm list 查看所有 storm 任務
> cd %STORM_HOME%
> storm ui
瀏覽器打開http://localhost:8080/ 可看到Storm運行。
下載download a pre-built jar。
部署這個jar在本地:
> storm jar storm-starter-0.0.1-SNAPSHOT-jar-with-dependencies.jar storm.starter.WordCountTopology WordCount -c nimbus.host=localhost
如果你刷新 Storm UI頁面,會看到 “WordCount” topology顯示列出,點按鏈接確認它處理數(shù)據(jù)。
storm 求 csdn 密碼庫中密碼出現(xiàn)的 topN,并直接在 eclipse 中調(diào)試運行:
package com.bj.test.top10; /** * @Author:tester * @DateTime:2016年6月21日 下午7:58:45 * @Description: Spout作為數(shù)據(jù)源,它實現(xiàn)了IRichSpout接口,功能是讀取一個文本文件并把它的每一行內(nèi)容發(fā)送給bolt。 * @Version:1.0 */ 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.OutputFieldsDeclarer; import backtype.storm.topology.base.BaseRichSpout; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Values; public class PasswdSpout extends BaseRichSpout { private SpoutOutputCollector collector; private FileReader fileReader; private boolean completed = false; public void ack(Object msgId) { System.out.println("==============OK:" + msgId); } public void close() { } public void fail(Object msgId) { System.out.println("++++++++++++++FAIL:" + msgId); } /** * 這是Spout最主要的方法,在這里我們讀取文本文件,并把它的每一行發(fā)射出去(給bolt) * 這個方法會不斷被調(diào)用,為了降低它對CPU的消耗,當任務完成時讓它sleep一下 * **/ public void nextTuple() { /** * The nextuple it is called forever, so if we have been readed the file * we will wait and then return */ if (completed) { try { Thread.sleep(1000); } catch (InterruptedException e) { // Do nothing } return; } String line; // Open the reader BufferedReader reader = new BufferedReader(fileReader); try { // Read all lines while ((line = reader.readLine()) != null) { String[] words = line.split("#"); String passwd = words[1].trim(); // Emit the word collector.emit(new Values(passwd)); /*for(String word : words){ word = word.trim(); if(!word.isEmpty()){ word = word.toLowerCase(); // Emit the word collector.emit(new Values(word)); } }*/ } } catch (Exception e) { throw new RuntimeException("Error reading tuple", e); } finally { completed = true; } } /** * 這是第一個方法,里面接收了三個參數(shù),第一個是創(chuàng)建Topology時的配置, * 第二個是所有的Topology數(shù)據(jù),第三個是用來把Spout的數(shù)據(jù)發(fā)射給bolt * **/ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { try { //獲取創(chuàng)建Topology時指定的要讀取的文件路徑 this.fileReader = new FileReader(conf.get("wordsFile").toString()); } catch (FileNotFoundException e) { throw new RuntimeException("Error reading file [" + conf.get("wordFile") + "]"); } //初始化發(fā)射器 this.collector = collector; } /** * Declare the output field "word" */ public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word")); } } ///////////////////////////////////////////////////////////////////////////////////////////// package com.bj.test.top10; import java.util.HashMap; import java.util.Map; import java.util.NavigableMap; import java.util.TreeMap; import backtype.storm.task.TopologyContext; import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.base.BaseBasicBolt; import backtype.storm.tuple.Tuple; import static com.bj.test.top10.SortMapByValue.*; public class Top10Bolt extends BaseBasicBolt { Integer id; String name; NavigableMap<String, Integer> counters; /** * Topology執(zhí)行完畢的清理工作,比如關(guān)閉連接、釋放資源等操作都會寫在這里 * 因為這只是個Demo,我們用它來打印我們的計數(shù)器 * */ @Override public void cleanup() { System.out.println(">>>>>>>>>>>> Word Counter ["+name+"-"+id+"] <<<<<<<<<<<"); /*for(Map.Entry<String, Integer> entry : counters.entrySet()){ System.out.println(entry.getKey()+": "+entry.getValue()); }*/ printMap(list2Map(sortMapByValuesTopN(counters, 10))); } /** * On create */ @Override public void prepare(Map stormConf, TopologyContext context) { this.counters = new TreeMap<String, Integer>().descendingMap(); this.name = context.getThisComponentId(); this.id = context.getThisTaskId(); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) {} // Bolt中最重要的是execute方法,每當一個tuple傳過來時它便會被調(diào)用 @Override public void execute(Tuple input, BasicOutputCollector collector) { String word = input.getString(0); /** * If the word dosn't exist in the map we will create * this, if not We will add 1 */ if(!counters.containsKey(word)){ counters.put(word, 1); }else{ Integer count = counters.get(word) + 1; counters.put(word, count); } } } ///////////////////////////////////////////////////////////////////////////////////////////// package com.bj.test.top10; /** * @Author:tester * @DateTime:2016年6月21日 下午7:52:32 * @Description: * @Version:1.0 */ import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.topology.TopologyBuilder; import backtype.storm.tuple.Fields; public class TopologyMain { public static void main(String[] args) throws InterruptedException { // 定義一個Topology TopologyBuilder builder = new TopologyBuilder(); // executor的數(shù)目, set parallelism hint to 4 builder.setSpout("PasswdSpout", new PasswdSpout(), 1); // set tasks number to 4 builder.setBolt("Top10Bolt", new Top10Bolt(), 1).setNumTasks(1).fieldsGrouping("PasswdSpout", new Fields("word")); // 配置 Config conf = new Config(); conf.put("wordsFile", "H:\\MySQL\\csdn_database\\www.csdn.net.100.sql"); // conf.put("wordsFile", "H:\\mysql\\csdn_database\\www.csdn.net.sql"); conf.setDebug(false); conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1); // use two worker processes // conf.setNumWorkers(4); // 創(chuàng)建一個本地模式cluster LocalCluster cluster = new LocalCluster(); // 提交Topology cluster.submitTopology("Getting-Started-Toplogie", conf, builder.createTopology()); Thread.sleep(1000); cluster.shutdown(); } }
關(guān)于“windows如何安裝storm eclipse調(diào)試TopN實例”這篇文章就分享到這里了,希望以上內(nèi)容可以對大家有一定的幫助,使各位可以學到更多知識,如果覺得文章不錯,請把它分享出去讓更多的人看到。
文章標題:windows如何安裝stormeclipse調(diào)試TopN實例
當前路徑:http://m.rwnh.cn/article20/jdgcco.html
成都網(wǎng)站建設公司_創(chuàng)新互聯(lián),為您提供面包屑導航、網(wǎng)站設計、商城網(wǎng)站、外貿(mào)網(wǎng)站建設、全網(wǎng)營銷推廣、網(wǎng)站設計公司
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)