计数
直接上代码
public class LocalStormSumTopology {
public static void main(String[] agrs) {
//Topology是通过build模式创建出来的
//storm中的所有作业都是通过topology来指定的
TopologyBuilder builder = new TopologyBuilder();
//在设置bolt到topology时,需要设置该bolt的上游的spout或者bolt的id,这样topology才知道该bolt的执行顺序,有点类似于单向链表结构,
//每一个环节持有上一个环节的引用,在bolt这里是持有上一个环节的id,这样同样可以定位到上一个环节
builder.setSpout("DataSourceSpout", new DataSourceSpout());
builder.setBolt("TotalBolt", new TotalBolt()).shuffleGrouping("DataSourceSpout");
//启动一个本地的Storm集群,不需要搭真正的集群,本地集群使用LocalCluster来提交topology,如果是在生产环境上提交topology,那么使用
//这个类StormSubmitter来代替LocalCluster来提交topology
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("LocalStormSumTopology", new Config(), builder.createTopology());
}
private static final String NUM = "num";
/**
* 发送数据源的spout类,一般是继承BaseRichSpout这个类
*/
public static class DataSourceSpout extends BaseRichSpout {
private SpoutOutputCollector mCollector;
int num;
/**
* 在storm开始的开始工作前回调一次,在这里做初始化
*
* @param conf 配置参数
* @param context 上下文
* @param collector 数据发射器,用来将数据发送到bolt中,类似于rxjava的数据发射器
*/
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.mCollector = collector;
}
/**
* 这是一个死循环方法,会自动循环调用,这个方法用来发送数据到下游
*/
public void nextTuple() {
//将数据发射到bolt中,一般使用Values这个类,传入的是可变参数,底层封装成ArrayList
mCollector.emit(new Values(++num));
System.out.println("从spout发射出的数据:" + num);
Utils.sleep(1000);
}
/**
* 声明从spout中发射的数据的字段名,在bolt阶段可以通过这里预设置的字段名进行取值,类似于安卓中的使用sp传输,
* 字段名和发送出来的数据一一对应,这样如果下游需要接收多个数据发射源,那么可以通过该字段名来做区别
*
* @param declarer
*/
public void declareOutputFields(OutputFieldsDeclarer declarer) {
//一般使用Fields来进行封装字段名fields底层封装了ArrayList<String>
declarer.declare(new Fields(NUM));
}
@Override
public void close() {
this.mCollector = null;
}
}
public static class TotalBolt extends BaseRichBolt {
private int sum = 0;
/**
* 初始化方法,跟spout中的open方法类似,只会调用一次,在这里做初始化
*
* @param stormConf
* @param context
* @param collector
*/
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
}
/**
* 每从上游接收到一个数据,就调用该方法回调过来
*
* @param input 用来提取上一个流程传过来的数据
*/
public void execute(Tuple input) {
//通过在上游设置的字段名来获取数据
Integer integerByField = input.getIntegerByField(NUM);
sum += integerByField;
System.out.println("累加的结果是:" + sum);
}
/**
* 为往下游发送的数据加上字段名,方面区别数据的来源
* @param declarer
*/
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
}
}
词频统计
直接上代码
public class LocalWorldCountStormTopology {
public static void main(String[] agrs) {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("DataSourceSpout", new DataSourceSpout());
builder.setBolt("CountBolt", new CountBolt()).shuffleGrouping("DataSourceSpout");
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("LocalWorldCountStormTopology", new Config(), builder.createTopology());
}
/**
* 输出每一行文本的spout
*/
public static class DataSourceSpout extends BaseRichSpout {
private SpoutOutputCollector mCollector;
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.mCollector = collector;
}
public void nextTuple() {
//通过这个方法,可以获取到某一个文件夹下所有符合规定后缀的文件,并且可以设置是否递归获取
Collection<File> files = FileUtils.listFiles(new File("/Users/teng/Downloads"), new String[]{"txt"}, true);
try {
for (File file : files) {
//因为下一步还需要做切割,因此需要先将文件一行一行取出来,放在String集合中
List<String> lines = FileUtils.readLines(file);
for (String line : lines) {
//使用,进行分割
String[] split = line.split(",");
//发射单词出去
for (String s : split) {
mCollector.emit(new Values(s));
}
}
//执行完成一次之后,需要修改文件名,这样就不用一直执行
FileUtils.moveFile(file, new File(file.getAbsolutePath()+System.currentTimeMillis()));
}
} catch (IOException e) {
e.printStackTrace();
}
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
//定义数据的字段名
declarer.declare(new Fields("word"));
}
}
/**
* 统计词频的bolt
*/
public static class CountBolt extends BaseRichBolt {
private Map<String, Integer> map = new HashMap<String, Integer>();
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
}
public void execute(Tuple input) {
String word = input.getStringByField("word");
Integer num = map.get(word);
if (num == null) {
num = 1;
} else {
num++;
}
map.put(word, num);
System.out.println("~~~~~~~~~");
Set<Map.Entry<String, Integer>> entries = map.entrySet();
for (Map.Entry<String, Integer> entry : entries) {
System.out.println(entry.getKey() + "出现的次数为:" + entry.getValue());
}
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
}
}