Storm如何实现单词计数
这篇文章主要介绍“Storm如何实现单词计数”，在日常操作中，相信很多人在Storm如何实现单词计数问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”Storm如何实现单词计数”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！
为雄县等地区用户提供了全套网页设计制作服务，及雄县网站建设行业解决方案。主营业务为网站建设、成都网站建设、雄县网站设计，以传统方式定制建设网站，并提供域名空间备案等一条龙服务，秉承以专业、用心的态度为用户提供真诚的服务。我们深信只要达到每一位用户的要求，就会得到认可，从而选择与我们长期合作。这样，我们也可以走得更远！
1. 使用mvn命令创建项目
mvn archetype:generate -DgroupId=storm.test -DartifactId=Storm01 -DpackageName=com.zhch.v1
然后编辑配置文件pom.xml，添加storm依赖
org.apache.storm storm-core 0.9.4
最后通过下述命令来编译项目，编译正确完成后导入到IDE中
mvn install
当然，也可以在IDE中安装maven插件，从而直接在IDE中创建maven项目
2. 实现数据源，用重复的静态语句来模拟数据源
package storm.test.v1;
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.util.Map;
public class SentenceSpout extends BaseRichSpout {
private String[] sentences = {
"storm integrates with the queueing",
"and database technologies you already use",
"a storm topology consumes streams of data",
"and processes those streams in arbitrarily complex ways",
"repartitioning the streams between each stage of the computation however needed"
};
private int index = 0;
private SpoutOutputCollector collector;
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
outputFieldsDeclarer.declare(new Fields("sentence"));
}
@Override
public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
this.collector = spoutOutputCollector;
}
@Override
public void nextTuple() {
this.collector.emit(new Values(sentences[index]));
index++;
if (index >= sentences.length) {
index = 0;
}
try {
Thread.sleep(1);
} catch (InterruptedException e) {
}
}
}3. 实现语句分割bolt
package storm.test.v1;
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;
@Override
public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
this.collector = outputCollector;
}
@Override
public void execute(Tuple tuple) {
String sentence = tuple.getStringByField("sentence");
String[] words = sentence.split(" ");
for (String word : words) {
this.collector.emit(new Values(word));
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
outputFieldsDeclarer.declare(new Fields("word"));
}
}4. 实现单词计数bolt
package storm.test.v1;
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.HashMap;
import java.util.Map;
public class WordCountBolt extends BaseRichBolt {
private OutputCollector collector;
private HashMap counts = null;
@Override
public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
this.collector = outputCollector;
this.counts = new HashMap();
}
@Override
public 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);
this.collector.emit(new Values(word, count));
}
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
outputFieldsDeclarer.declare(new Fields("word", "count"));
}
} 5. 实现上报bolt
package storm.test.v1;
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.Tuple;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class ReportBolt extends BaseRichBolt {
private HashMap counts = null;
@Override
public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
counts = new HashMap();
}
@Override
public void execute(Tuple tuple) {
String word = tuple.getStringByField("word");
Long count = tuple.getLongByField("count");
this.counts.put(word, count);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
}
@Override
public void cleanup() { //本地模式下，终止topology时可以保证cleanup()被执行
System.out.println("--- FINAL COUNTS ---");
List keys = new ArrayList();
keys.addAll(this.counts.keySet());
Collections.sort(keys);
for (String key : keys) {
System.out.println(key + " : " + this.counts.get(key));
}
System.out.println("----------");
}
} 6. 实现单词计数topology
package storm.test.v1;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
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 REPORT_BOLT_ID = "report-bolt";
private static final String TOPOLOGY_NAME = "word-count-topology";
public static void main(String[] args) {
SentenceSpout spout = new SentenceSpout();
SplitSentenceBolt spiltBolt = new SplitSentenceBolt();
WordCountBolt countBolt = new WordCountBolt();
ReportBolt reportBolt = new ReportBolt();
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(SENTENCE_SPOUT_ID, spout); //注册数据源
builder.setBolt(SPLIT_BOLT_ID, spiltBolt) //注册bolt
.shuffleGrouping(SENTENCE_SPOUT_ID); //该bolt订阅spout随机均匀发射来的数据流
builder.setBolt(COUNT_BOLT_ID, countBolt)
.fieldsGrouping(SPLIT_BOLT_ID, new Fields("word")); //该bolt订阅spiltBolt发射来的数据流，并且保证"word"字段值相同的tuple会被路由到同一个countBolt
builder.setBolt(REPORT_BOLT_ID, reportBolt)
.globalGrouping(COUNT_BOLT_ID); //该bolt订阅countBolt发射来的数据流，并且所有的tuple都会被路由到唯一的一个reportBolt中
Config config = new Config();
//本地模式启动
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();
}
}7. 运行结果：
--- FINAL COUNTS --- a : 302 already : 302 and : 604 arbitrarily : 302 between : 302 complex : 302 computation : 302 consumes : 302 data : 302 database : 302 each : 302 however : 302 in : 302 integrates : 302 needed : 302 of : 604 processes : 302 queueing : 302 repartitioning : 302 stage : 302 storm : 604 streams : 906 technologies : 302 the : 906 those : 302 topology : 302 use : 302 ways : 302 with : 302 you : 302 ----------
到此，关于“Storm如何实现单词计数”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注创新互联网站，小编会继续努力为大家带来更多实用的文章！
分享名称：Storm如何实现单词计数
标题路径：http://myzitong.com/article/ipehph.html