Skip to content

Commit 7f656e5

Browse files
author
centmeng
committed
WordCount统计频率例子和Message发送消息并实时处理例子
1 parent d04e2f3 commit 7f656e5

10 files changed

Lines changed: 610 additions & 0 deletions

File tree

‎src/com/msj/storm/bolt/SpliterBolt.java‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
package com.msj.storm.bolt;
2+
3+
import backtype.storm.task.OutputCollector;
4+
import backtype.storm.task.TopologyContext;
5+
import backtype.storm.topology.IRichBolt;
6+
import backtype.storm.topology.OutputFieldsDeclarer;
7+
import backtype.storm.tuple.Fields;
8+
import backtype.storm.tuple.Tuple;
9+
import backtype.storm.tuple.Values;
10+
11+
import java.util.Map;
12+
13+
/**
14+
* @author Vincent.M mengshaojie@188.com
15+
* @date 2017/5/11 下午2:57
16+
* @copyright ©2017 孟少杰 All Rights Reserved
17+
* @desc 传递消息并且实时处理-单词分割
18+
*/
19+
public class SpliterBolt implements IRichBolt {
20+
21+
private static final long serialVersionUID = 1L;
22+
23+
private OutputCollector collector;
24+
25+
@Override
26+
public void prepare(Map config, TopologyContext context, OutputCollector collector) {
27+
this.collector = collector;
28+
}
29+
30+
31+
private boolean flag = false;
32+
33+
@Override
34+
public void execute(Tuple tuple) {
35+
try {
36+
String subjects = tuple.getStringByField("subjects");
37+
38+
// if(!flag && subjects.equals("flume,activiti")){
39+
// flag = true;
40+
// int a = 1/0;
41+
// }
42+
43+
String[] words = subjects.split(",");
44+
//List<String> list = new ArrayList<String>();
45+
//int index = 0;
46+
for (String word : words) {
47+
//注意这里循环发送消息，要携带tuple对象，用于处理异常时重发策略
48+
collector.emit(tuple, new Values(word));
49+
//list.add(word);
50+
//index ++;
51+
}
52+
//collector.emit(tuple, new Values(list));
53+
collector.ack(tuple);
54+
} catch (Exception e) {
55+
e.printStackTrace();
56+
collector.fail(tuple);
57+
}
58+
}
59+
60+
@Override
61+
public void declareOutputFields(OutputFieldsDeclarer declarer) {
62+
declarer.declare(new Fields("word"));
63+
}
64+
65+
@Override
66+
public void cleanup() {
67+
68+
}
69+
70+
@Override
71+
public Map<String, Object> getComponentConfiguration() {
72+
return null;
73+
}
74+
75+
}

‎src/com/msj/storm/bolt/WordCountBolt.java‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
package com.msj.storm.bolt;
2+
3+
import backtype.storm.task.OutputCollector;
4+
import backtype.storm.task.TopologyContext;
5+
import backtype.storm.topology.IRichBolt;
6+
import backtype.storm.topology.OutputFieldsDeclarer;
7+
import backtype.storm.tuple.Fields;
8+
import backtype.storm.tuple.Tuple;
9+
import backtype.storm.tuple.Values;
10+
11+
import java.util.HashMap;
12+
import java.util.Map;
13+
14+
/**
15+
* @author Vincent.M mengshaojie@188.com
16+
* @date 2017/5/11 下午8:20
17+
* @copyright ©2017 孟少杰 All Rights Reserved
18+
* @desc 查看出现频率-计数统计bolt
19+
*/
20+
public class WordCountBolt implements IRichBolt {
21+
22+
private OutputCollector collector;
23+
private HashMap<String, Long> counts = null;
24+
25+
@Override
26+
public void prepare(Map config, TopologyContext context, OutputCollector collector) {
27+
this.collector = collector;
28+
this.counts = new HashMap<String, Long>();
29+
}
30+
31+
@Override
32+
public void execute(Tuple tuple) {
33+
String word = tuple.getStringByField("word");
34+
Long count = this.counts.get(word);
35+
if(count == null){
36+
count = 0L;
37+
}
38+
count++;
39+
this.counts.put(word, count);
40+
this.collector.emit(new Values(word, count));
41+
}
42+
43+
@Override
44+
public void declareOutputFields(OutputFieldsDeclarer declarer) {
45+
declarer.declare(new Fields("word", "count"));
46+
}
47+
48+
@Override
49+
public Map<String, Object> getComponentConfiguration() {
50+
return null;
51+
}
52+
53+
@Override
54+
public void cleanup() {
55+
56+
}
57+
58+
}

‎src/com/msj/storm/bolt/WordReportBolt.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
package com.msj.storm.bolt;
2+
3+
import backtype.storm.task.OutputCollector;
4+
import backtype.storm.task.TopologyContext;
5+
import backtype.storm.topology.IRichBolt;
6+
import backtype.storm.topology.OutputFieldsDeclarer;
7+
import backtype.storm.tuple.Tuple;
8+
9+
import java.util.*;
10+
11+
/**
12+
* @author Vincent.M mengshaojie@188.com
13+
* @date 2017/5/11 下午8:20
14+
* @copyright ©2017 孟少杰 All Rights Reserved
15+
* @desc 查看出现频率-记录bolt
16+
*/
17+
public class WordReportBolt implements IRichBolt {
18+
19+
private HashMap<String, Long> counts = null;
20+
21+
@Override
22+
public void prepare(Map config, TopologyContext context, OutputCollector collector) {
23+
this.counts = new HashMap<String, Long>();
24+
}
25+
26+
@Override
27+
public void execute(Tuple tuple) {
28+
String word = tuple.getStringByField("word");
29+
Long count = tuple.getLongByField("count");
30+
this.counts.put(word, count);
31+
}
32+
33+
@Override
34+
public void declareOutputFields(OutputFieldsDeclarer declarer) {
35+
36+
}
37+
38+
@Override
39+
public Map<String, Object> getComponentConfiguration() {
40+
return null;
41+
}
42+
43+
@Override
44+
public void cleanup() {
45+
System.out.println("----------- FINAL COUNTS -----------");
46+
List<String> keys = new ArrayList<String>();
47+
keys.addAll(this.counts.keySet());
48+
Collections.sort(keys);
49+
for (String key : keys) {
50+
System.out.println(key + " : " + this.counts.get(key));
51+
}
52+
System.out.println("-----------------------------------");
53+
}
54+
55+
}

‎src/com/msj/storm/bolt/WordSplitBolt.java‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
package com.msj.storm.bolt;
2+
3+
import backtype.storm.task.OutputCollector;
4+
import backtype.storm.task.TopologyContext;
5+
import backtype.storm.topology.IRichBolt;
6+
import backtype.storm.topology.OutputFieldsDeclarer;
7+
import backtype.storm.tuple.Fields;
8+
import backtype.storm.tuple.Tuple;
9+
import backtype.storm.tuple.Values;
10+
11+
import java.util.Map;
12+
13+
/**
14+
* @author Vincent.M mengshaojie@188.com
15+
* @date 2017/5/11 下午8:20
16+
* @copyright ©2017 孟少杰 All Rights Reserved
17+
* @desc 查看出现频率-分割bolt
18+
*/
19+
public class WordSplitBolt implements IRichBolt{
20+
21+
private static final String ERROR_STR = "don't have a cow man";
22+
23+
private OutputCollector collector;
24+
25+
@Override
26+
public void prepare(Map config, TopologyContext context, OutputCollector collector) {
27+
this.collector = collector;
28+
}
29+
30+
31+
@Override
32+
public void execute(Tuple tuple) {
33+
String sentence = tuple.getStringByField("sentence");
34+
String[] words = sentence.split(" ");
35+
for(String word : words){
36+
this.collector.emit(new Values(word));
37+
//this.collector.ack(tuple);
38+
}
39+
}
40+
41+
42+
@Override
43+
public void declareOutputFields(OutputFieldsDeclarer declarer) {
44+
declarer.declare(new Fields("word"));
45+
}
46+
47+
@Override
48+
public void cleanup() {
49+
50+
}
51+
52+
@Override
53+
public Map<String, Object> getComponentConfiguration() {
54+
return null;
55+
}
56+
57+
58+
}

‎src/com/msj/storm/bolt/WriterBolt.java‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
package com.msj.storm.bolt;
2+
3+
import backtype.storm.task.OutputCollector;
4+
import backtype.storm.task.TopologyContext;
5+
import backtype.storm.topology.IRichBolt;
6+
import backtype.storm.topology.OutputFieldsDeclarer;
7+
import backtype.storm.tuple.Tuple;
8+
import backtype.storm.tuple.Values;
9+
10+
import java.io.FileWriter;
11+
import java.io.IOException;
12+
import java.util.Map;
13+
14+
/**
15+
* @author Vincent.M mengshaojie@188.com
16+
* @date 2017/5/11 下午2:57
17+
* @copyright ©2017 孟少杰 All Rights Reserved
18+
* @desc 传递消息并且实时处理-输出写入bolt
19+
*/
20+
public class WriterBolt implements IRichBolt {
21+
22+
private static final long serialVersionUID = 1L;
23+
24+
private FileWriter writer;
25+
26+
private OutputCollector collector;
27+
28+
@Override
29+
public void prepare(Map config, TopologyContext context, OutputCollector collector) {
30+
this.collector = collector;
31+
try {
32+
writer = new FileWriter("d://message.txt");
33+
} catch (IOException e) {
34+
e.printStackTrace();
35+
}
36+
}
37+
38+
private boolean flag = false;
39+
40+
@Override
41+
public void execute(Tuple tuple) {
42+
String word = tuple.getString(0);
43+
// List<String> list = (List<String>)tuple.getValueByField("word");
44+
// System.out.println("======================" + list);
45+
try {
46+
if(!flag && word.equals("hadoop")){
47+
flag = true;
48+
int a = 1/0;
49+
}
50+
writer.write(word);
51+
writer.write("\r\n");
52+
writer.flush();
53+
} catch (Exception e) {
54+
e.printStackTrace();
55+
collector.fail(tuple);
56+
}
57+
collector.emit(tuple, new Values(word));
58+
collector.ack(tuple);
59+
}
60+
61+
@Override
62+
public void cleanup() {
63+
64+
}
65+
66+
@Override
67+
public void declareOutputFields(OutputFieldsDeclarer declarer) {
68+
69+
}
70+
71+
@Override
72+
public Map<String, Object> getComponentConfiguration() {
73+
return null;
74+
}
75+
76+
}

0 commit comments

Comments
 (0)