storm单词计数 本地运行
2017-06-20 16:44
281 查看
import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.commons.io.FileUtils;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import cn.crxy.storm.LocalStormTopology.SumBolt;
public class WordcountStormTopology {
public static class DataSourceSpout extends BaseRichSpout{
private Map conf;
private TopologyContext context;
private SpoutOutputCollector collector;
/**
* 在本实例执行的时候被调用一次
*/
public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
this.conf = conf;
this.context = context;
this.collector = collector;
}
/**
* 死循环调用 心跳
*/
public void nextTuple() {
//获取指定目录以下全部的文件
Collection<File> files = FileUtils.listFiles(new File("D:\\test"), new String[]{"txt"}, true);
for (File file : files) {
try {
//解析每个文件的每一行
List<String> readLines = FileUtils.readLines(file);
for (String line : readLines) {
//把每一行数据发送出去
this.collector.emit(new Values(line));
}
//重命名 防止多次读
FileUtils.moveFile(file, new File(file.getAbsolutePath()+System.currentTimeMillis()));
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 声明字段名称
*/
public void declareOutputFields(OutputFieldsDeclarer declarer) {
//fields就是field的列表
declarer.declare(new Fields("line"));
}
}
public static class SpiltBolt extends BaseRichBolt{
private Map stormConf;
private TopologyContext context;
private OutputCollector collector;
/**
* 仅仅会被调用一次
*/
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.stormConf = stormConf;
this.context = context;
this.collector = collector;
}
/**
* 死循环,循环的获取上一级发送过来的数据(spout/bolt)
*/
public void execute(Tuple input) {
//获取tuple发来数据
String line = input.getStringByField("line");
//对每一行数据进行分割
String[] words = line.split("\t");
for (String word : words) {
//把分割的单词发送到下一个bolt
this.collector.emit(new Values(word));
}
}
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.commons.io.FileUtils;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import cn.crxy.storm.LocalStormTopology.SumBolt;
public class WordcountStormTopology {
public static class DataSourceSpout extends BaseRichSpout{
private Map conf;
private TopologyContext context;
private SpoutOutputCollector collector;
/**
* 在本实例执行的时候被调用一次
*/
public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
this.conf = conf;
this.context = context;
this.collector = collector;
}
/**
* 死循环调用 心跳
*/
public void nextTuple() {
//获取指定目录以下全部的文件
Collection<File> files = FileUtils.listFiles(new File("D:\\test"), new String[]{"txt"}, true);
for (File file : files) {
try {
//解析每个文件的每一行
List<String> readLines = FileUtils.readLines(file);
for (String line : readLines) {
//把每一行数据发送出去
this.collector.emit(new Values(line));
}
//重命名 防止多次读
FileUtils.moveFile(file, new File(file.getAbsolutePath()+System.currentTimeMillis()));
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 声明字段名称
*/
public void declareOutputFields(OutputFieldsDeclarer declarer) {
//fields就是field的列表
declarer.declare(new Fields("line"));
}
}
public static class SpiltBolt extends BaseRichBolt{
private Map stormConf;
private TopologyContext context;
private OutputCollector collector;
/**
* 仅仅会被调用一次
*/
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.stormConf = stormConf;
this.context = context;
this.collector = collector;
}
/**
* 死循环,循环的获取上一级发送过来的数据(spout/bolt)
*/
public void execute(Tuple input) {
//获取tuple发来数据
String line = input.getStringByField("line");
//对每一行数据进行分割
String[] words = line.split("\t");
for (String word : words) {
//把分割的单词发送到下一个bolt
this.collector.emit(new Values(word));
}
}
相关文章推荐
- 005-采用storm程序对目录下文件的单词计数本地模式运行
- storm两个案例(1单词计数本地执行 2累加集群执行 3集群关闭storm任务写法)
- storm单词计数 本地执行
- 004-storm开发计数程序本地模式运行
- twitter storm安装和storm-start的本地运行
- storm本地模式运行Local mode
- 本地运行storm时报错
- centos的twitter storm安装和storm-start的本地运行
- 【Storm】storm安装、配置、使用以及Storm单词计数程序的实例分析
- JStorm/Storm的调试:本地运行模式
- 007-storm开发计数程序集群模式运行
- Storm实验 -- 单词计数3
- strom-1.1.0模拟单词统计功能,Spout编写,Bolt编写,TopologyDriver编写,本地模式运行,集群模式运行,集群模式下看输出结果
- 玩转linux主机--hadoop伪分布式配置及单词计数程序的运行
- Storm实验 -- 单词计数4
- Storm实现单词计数
- twitter storm安装和storm-start的本地运行
- 【Storm】storm安装、配置、使用以及Storm单词计数程序的实例分析
- Storm官方提供的trident单词计数的例子
- Storm 1.0.2 - 单词计数案例学习