Storm实验 -- 单词计数2
2019独角兽企业重金招聘Python工程师标准>>>
在上一次单词计数的基础上做如下改动:
1. 使用可靠的消息处理机制
2. 配置 worker 、executor、task 数量
3. 使用集群模式提交
数据源spout
package com.zhch.v2;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;import java.io.BufferedReader;
import java.io.FileReader;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;public class SentenceSpout extends BaseRichSpout {private FileReader fileReader = null;private boolean completed = false;private ConcurrentHashMap<UUID, Values> pending;private SpoutOutputCollector collector;@Overridepublic void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {outputFieldsDeclarer.declare(new Fields("sentence"));}@Overridepublic void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {this.collector = spoutOutputCollector;this.pending = new ConcurrentHashMap<UUID, Values>();try {this.fileReader = new FileReader(map.get("wordsFile").toString());} catch (Exception e) {throw new RuntimeException("Error reading file [" + map.get("wordsFile") + "]");}}@Overridepublic void nextTuple() {if (completed) {try {Thread.sleep(1000);} catch (InterruptedException e) {}}String line;BufferedReader reader = new BufferedReader(fileReader);try {while ((line = reader.readLine()) != null) {Values values = new Values(line);UUID msgId = UUID.randomUUID();this.pending.put(msgId, values);this.collector.emit(values, msgId); //发射时带上msgId,实现可靠的消息处理}} catch (Exception e) {throw new RuntimeException("Error reading tuple", e);} finally {completed = true;}}@Overridepublic void ack(Object msgId) {this.pending.remove(msgId); //确认时,从列表中删除该tuple}@Overridepublic void fail(Object msgId) {this.collector.emit(this.pending.get(msgId), msgId); //失败或超时,重新发射}
}
实现语句分割bolt
package com.zhch.v2;import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;import java.util.Map;public class SplitSentenceBolt extends BaseRichBolt {private OutputCollector collector;@Overridepublic void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {this.collector = outputCollector;}@Overridepublic void execute(Tuple tuple) {String sentence = tuple.getStringByField("sentence");String[] words = sentence.split(" ");for (String word : words) {this.collector.emit(tuple, new Values(word)); //将输出的tuple和输入的tuple锚定,实现可靠的消息处理}this.collector.ack(tuple);}@Overridepublic void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {outputFieldsDeclarer.declare(new Fields("word"));}
}
实现单词计数bolt
package com.zhch.v2;import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;import java.io.BufferedWriter;
import java.io.FileWriter;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;public class WordCountBolt extends BaseRichBolt {private OutputCollector collector;private HashMap<String, Long> counts = null;@Overridepublic void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {this.collector = outputCollector;this.counts = new HashMap<String, Long>();}@Overridepublic void execute(Tuple tuple) {String word = tuple.getStringByField("word");Long count = this.counts.get(word);if (count == null) {count = 0L;}count++;this.counts.put(word, count);BufferedWriter writer = null;try {writer = new BufferedWriter(new FileWriter("/home/grid/stormData/result.txt"));Iterator<String> keys = this.counts.keySet().iterator();while (keys.hasNext()) {String w = keys.next();Long c = this.counts.get(w);writer.write(w + " : " + c);writer.newLine();writer.flush();}} catch (Exception e) {e.printStackTrace();} finally {if (writer != null) {try {writer.close();} catch (Exception e) {e.printStackTrace();}writer = null;}}this.collector.ack(tuple);}@Overridepublic void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {outputFieldsDeclarer.declare(new Fields("word", "count"));}
}
实现单词计数topology
package com.zhch.v2;import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;public class WordCountTopology {private static final String SENTENCE_SPOUT_ID = "sentence-spout";private static final String SPLIT_BOLT_ID = "split-bolt";private static final String COUNT_BOLT_ID = "count-bolt";private static final String TOPOLOGY_NAME = "word-count-topology-v2";public static void main(String[] args) throws Exception {SentenceSpout spout = new SentenceSpout();SplitSentenceBolt spiltBolt = new SplitSentenceBolt();WordCountBolt countBolt = new WordCountBolt();TopologyBuilder builder = new TopologyBuilder();builder.setSpout(SENTENCE_SPOUT_ID, spout, 2); //使用2个spout executorbuilder.setBolt(SPLIT_BOLT_ID, spiltBolt, 2).setNumTasks(4) //使用2个spiltBolt executor,4个task.shuffleGrouping(SENTENCE_SPOUT_ID);builder.setBolt(COUNT_BOLT_ID, countBolt, 2) //使用2个countBolt executor.fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));Config config = new Config();config.put("wordsFile", args[0]);if (args != null && args.length > 1) {config.setNumWorkers(2); //使用2个worker processes//集群模式启动StormSubmitter.submitTopology(args[1], config, builder.createTopology());} else {LocalCluster cluster = new LocalCluster();cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());try {Thread.sleep(5 * 1000);} catch (InterruptedException e) {}cluster.killTopology(TOPOLOGY_NAME);cluster.shutdown();}}
}
编写完程序后使用命令 mvn clean install 打包,然后提交到Storm集群
[grid@hadoop4 stormTarget]$ storm jar Storm02-1.0-SNAPSHOT.jar com.zhch.v2.WordCountTopology /home/grid/stormData/input.txt word-count-topology-v2
运行结果:
[grid@hadoop5 stormData]$ cat result.txt
can : 1
second : 1
simple : 1
set : 1
data : 2
unbounded : 1
has : 1
Apache : 1
open : 1
over : 1
free : 1
easy : 2
fast: : 1
reliably : 1
any : 1
with : 1
million : 1
is : 6
learning : 1
analytics : 1
torm : 1
node : 1
processed : 2
what : 1
batch : 1
operate : 1
will : 1
language : 1
fault-tolerant : 1
[grid@hadoop6 stormData]$ cat result.txt
to : 3
for : 2
distributed : 2
use : 2
used : 1
Storm : 4
It : 1
online : 1
cases: : 1
of : 2
programming : 1
more : 1
clocked : 1
scalable : 1
processing : 2
guarantees : 1
be : 2
ETL : 1
continuous : 1
it : 2
Hadoop : 1
makes : 1
your : 1
a : 4
at : 1
did : 1
fun : 1
machine : 1
up : 1
and : 5
process : 1
RPC : 1
many : 1
system : 1
source : 1
realtime : 3
benchmark : 1
per : 2
doing : 1
lot : 1
streams : 1
computation : 2
tuples : 1
转载于:https://my.oschina.net/zc741520/blog/410080
Storm实验 -- 单词计数2相关推荐
- Storm实验 -- 单词计数4
2019独角兽企业重金招聘Python工程师标准>>> 在上一次单词计数的基础上做如下改动: 使用 自定义 分组策略,将首字母相同的单词发送给同一个task计数 自定义 Custo ...
- [Storm]分布式单词计数(一)一个简单的storm demo
目录 前言: 基本概念: 1.1 Spout 1.1 业务 SentenceSpout 1.2 SentenceSpout引用的部分类源码 BaseRichSpout源码 Values源码 ISpou ...
- 【Storm】storm安装、配置、使用以及Storm单词计数程序的实例分析
前言:阅读笔记 storm和hadoop集群非常像.hadoop执行mr.storm执行topologies. mr和topologies最关键的不同点是:mr执行终于会结束,而topologies永 ...
- 大数据【四】MapReduce(单词计数;二次排序;计数器;join;分布式缓存)
前言: 根据前面的几篇博客学习,现在可以进行MapReduce学习了.本篇博客首先阐述了MapReduce的概念及使用原理,其次直接从五个实验中实践学习(单词计数,二次排序,计数器,join,分 ...
- MIT自然语言处理第二讲:单词计数(第三、四部分)
MIT自然语言处理第二讲:单词计数(第三部分) 自然语言处理:单词计数 Natural Language Processing: (Simple) Word Counting 作者:Regina Ba ...
- MIT自然语言处理第二讲:单词计数(第一、二部分)
MIT自然语言处理第二讲:单词计数(第一部分) 自然语言处理:单词计数 Natural Language Processing: (Simple) Word Counting 作者:Regina Ba ...
- c语言字符统计2sdut,C语言实验——单词统计(字符串) SDUT
C语言实验--单词统计 Problem Description 从键盘输入一行字符(长度小于100),统计其中单词的个数,各单词以空格分隔,且空格数可以是多个. Input 输入只有一行句子.仅有空格 ...
- Spark Core:Scala单词计数
Spark Core:Scala单词计数 文章目录 Spark Core:Scala单词计数 1.实验描述 2.实验环境 3.相关技能 4.知识点 5.实验效果 6.实验步骤 8.总结 1.实验描述 ...
- 软件工程小组项目——单词计数
软件工程小组项目--单词计数 项目地址 PSP表格 解题思路 实现过程 1.字符数 2.单词数 3.行数 4.各段行数统计 5.对子目录文件进行操作 6.图形界面 性能优化 实验总结 项目地址 Git ...
- 自然语言处理第二讲:单词计数
自然语言处理:单词计数 这一讲主要内容(Today): 1.语料库及其性质: 2.Zipf 法则: 3.标注语料库例子: 4.分词算法: 一. 语料库及其性质: a) 什么是语料库(Corpora) ...
最新文章
- 初识Tcl(一):Tcl 命令
- java final 变量 回收_在Java中将final用于变量会改善垃圾回收吗?
- HDU1576(欧几里得算法)
- 计算机二级web题目(6)--动态网页技术概述
- python tkinter布局混用_python tkinter布局界面如何实现?
- 基于分数阶傅里叶变换的chirp信号检测与参数估计(原理附代码)
- 解决办法:My Endnote Library.enl constrains an incorrect path
- js表单提交的三种方式
- 计算某天是星期几-泽勒算法
- GitLab配置ssh key:gitlab add an ssh key
- [安装之1] dell设置从ssd启动_如何进bios设置ssd固态硬盘为第一启动
- Linux下的dd和cat
- 学会提问——批判性思维指南
- 一正云系统服务器搭建,腾正云服务器
- BootStrap 导航栏实现下滑消失,上滑出现
- 计算机毕业设计Python+django 宠物领养中心小程序(源码+系统+mysql数据库+Lw文档)
- 11. Nginx HTTPS
- 在几何画板中如何制作圆柱的侧面展开动画_几何画板如何制作圆柱体形成的过程动画...
- 三星获欧洲专利居业内之首 中兴首次进入前十
- 高中计算机学业水平考是c合格了吗,高中会考多少分才算合格