重庆分公司,新征程启航
为企业提供网站建设、域名注册、服务器等服务
这篇文章将为大家详细讲解有关windows如何安装storm eclipse调试TopN实例,小编觉得挺实用的,因此分享给大家做个参考,希望大家阅读完这篇文章后可以有所收获。
成都创新互联公司专注于通道网站建设服务及定制,我们拥有丰富的企业做网站经验。 热诚为您提供通道营销型网站建设,通道网站制作、通道网页设计、通道网站官网定制、微信平台小程序开发服务,打造通道网络公司原创品牌,更为您提供通道网站排名全网营销落地服务。
配置Java环境变量 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 版本都无关。
http://dl.dropboxusercontent.com/s/iglqz73chkul1tu/storm-0.9.1-incubating-SNAPSHOT-12182013.zip
配置Storm环境变量
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已经在安装JDK时手动配置了环境变量,而Python好像是默认自动就会配置好环境变量的,
我的Python目录下没有Scripts文件夹,暂时这样配置就可以了,不影响下面的使用。
Nimbus
注意2:
一定要在 STORM_HOME 目录下执行后续命令,否则会报错。
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显示列出,点按链接确认它处理数据。
storm 求 csdn 密码库中密码出现的 topN,并直接在 eclipse 中调试运行:
package com.bj.test.top10; /** * @Author:tester * @DateTime:2016年6月21日 下午7:58:45 * @Description: Spout作为数据源,它实现了IRichSpout接口,功能是读取一个文本文件并把它的每一行内容发送给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最主要的方法,在这里我们读取文本文件,并把它的每一行发射出去(给bolt) * 这个方法会不断被调用,为了降低它对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; } } /** * 这是第一个方法,里面接收了三个参数,第一个是创建Topology时的配置, * 第二个是所有的Topology数据,第三个是用来把Spout的数据发射给bolt * **/ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { try { //获取创建Topology时指定的要读取的文件路径 this.fileReader = new FileReader(conf.get("wordsFile").toString()); } catch (FileNotFoundException e) { throw new RuntimeException("Error reading file [" + conf.get("wordFile") + "]"); } //初始化发射器 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; NavigableMapcounters; /** * Topology执行完毕的清理工作,比如关闭连接、释放资源等操作都会写在这里 * 因为这只是个Demo,我们用它来打印我们的计数器 * */ @Override public void cleanup() { System.out.println(">>>>>>>>>>>> Word Counter ["+name+"-"+id+"] <<<<<<<<<<<"); /*for(Map.Entry 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 ().descendingMap(); this.name = context.getThisComponentId(); this.id = context.getThisTaskId(); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) {} // Bolt中最重要的是execute方法,每当一个tuple传过来时它便会被调用 @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的数目, 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); // 创建一个本地模式cluster LocalCluster cluster = new LocalCluster(); // 提交Topology cluster.submitTopology("Getting-Started-Toplogie", conf, builder.createTopology()); Thread.sleep(1000); cluster.shutdown(); } }
关于“windows如何安装storm eclipse调试TopN实例”这篇文章就分享到这里了,希望以上内容可以对大家有一定的帮助,使各位可以学到更多知识,如果觉得文章不错,请把它分享出去让更多的人看到。