Skip to content

Commit 702f500

Browse files
author
centmeng
committed
Storm Trident示例
1 parent 1e1de15 commit 702f500

8 files changed

Lines changed: 530 additions & 0 deletions

File tree

‎src/com/msj/storm/trident/example/TridentFilter.java‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
package com.msj.storm.trident.example;
2+
3+
import storm.trident.Stream;
4+
import storm.trident.TridentTopology;
5+
import storm.trident.operation.BaseFilter;
6+
import storm.trident.operation.BaseFunction;
7+
import storm.trident.operation.TridentCollector;
8+
import storm.trident.testing.FixedBatchSpout;
9+
import storm.trident.tuple.TridentTuple;
10+
import backtype.storm.Config;
11+
import backtype.storm.LocalCluster;
12+
import backtype.storm.StormSubmitter;
13+
import backtype.storm.generated.StormTopology;
14+
import backtype.storm.tuple.Fields;
15+
import backtype.storm.tuple.Values;
16+
17+
/**
18+
* @author Vincent.M mengshaojie@188.com
19+
* @date 2017/5/11 下午2:57
20+
* @copyright ©2017 孟少杰 All Rights Reserved
21+
* @desc 测试Trident过滤功能
22+
*/
23+
public class TridentFilter {
24+
25+
//继承BaseFilter类，重新isKeep方法
26+
public static class CheckFilter extends BaseFilter {
27+
@Override
28+
public boolean isKeep(TridentTuple tuple) {
29+
int a = tuple.getInteger(0);
30+
int b = tuple.getInteger(1);
31+
if((a + b) % 2 == 0){
32+
return true;
33+
}
34+
return false;
35+
}
36+
}
37+
38+
//继承BaseFunction类，重新execute方法
39+
public static class Result extends BaseFunction {
40+
@Override
41+
public void execute(TridentTuple tuple, TridentCollector collector) {
42+
//获取tuple输入内容
43+
Integer a = tuple.getIntegerByField("a");
44+
Integer b = tuple.getIntegerByField("b");
45+
Integer c = tuple.getIntegerByField("c");
46+
Integer d = tuple.getIntegerByField("d");
47+
System.out.println("a: "+ a + ", b: " + b + ", c: " + c + ", d: " + d);
48+
}
49+
}
50+
51+
public static StormTopology buildTopology() {
52+
TridentTopology topology = new TridentTopology();
53+
//设定数据源
54+
FixedBatchSpout spout = new FixedBatchSpout(
55+
new Fields("a", "b", "c", "d"), //声明输入的域字段为"a"、"b"、"c"、"d"
56+
4, //设置批处理大小为1
57+
//设置数据源内容
58+
//测试数据
59+
new Values(1, 4, 7, 10),
60+
new Values(1, 1, 3, 11),
61+
new Values(2, 2, 7, 1),
62+
new Values(2, 5, 7, 2));
63+
//指定是否循环
64+
spout.setCycle(false);
65+
//指定输入源spout
66+
Stream inputStream = topology.newStream("spout", spout);
67+
/**
68+
* 要实现流sqout - bolt的模式 在trident里是使用each来做的
69+
* each方法参数：
70+
* 1.输入数据源参数名称：subjects
71+
* 2.需要流转执行的function对象（也就是bolt对象）：new Split()
72+
*/
73+
inputStream.each(new Fields("a", "b", "c", "d"), new CheckFilter())
74+
//继续使用each调用下一个function（bolt）输入参数为subject和count，第二个参数为new Result() 也就是执行函数，第三个参数为没有输出
75+
.each(new Fields("a", "b", "c", "d"), new Result(), new Fields());
76+
return topology.build(); //利用这种方式，我们返回一个StormTopology对象，进行提交
77+
}
78+
79+
public static void main(String[] args) throws Exception {
80+
81+
Config conf = new Config();
82+
//设置batch最大处理
83+
conf.setNumWorkers(2);
84+
conf.setMaxSpoutPending(20);
85+
if(args.length == 0) {
86+
LocalCluster cluster = new LocalCluster();
87+
cluster.submitTopology("trident-function", conf, buildTopology());
88+
Thread.sleep(10000);
89+
cluster.shutdown();
90+
} else {
91+
StormSubmitter.submitTopology(args[0], conf, buildTopology());
92+
}
93+
94+
}
95+
}

‎src/com/msj/storm/trident/example/TridentFunction.java‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
package com.msj.storm.trident.example;
2+
3+
import storm.trident.Stream;
4+
import storm.trident.TridentTopology;
5+
import storm.trident.operation.BaseFunction;
6+
import storm.trident.operation.TridentCollector;
7+
import storm.trident.testing.FixedBatchSpout;
8+
import storm.trident.tuple.TridentTuple;
9+
import backtype.storm.Config;
10+
import backtype.storm.LocalCluster;
11+
import backtype.storm.StormSubmitter;
12+
import backtype.storm.generated.StormTopology;
13+
import backtype.storm.tuple.Fields;
14+
import backtype.storm.tuple.Values;
15+
16+
/**
17+
* @author Vincent.M mengshaojie@188.com
18+
* @date 2017/5/11 下午2:57
19+
* @copyright ©2017 孟少杰 All Rights Reserved
20+
* @desc 测试TridentFunction
21+
*/
22+
public class TridentFunction {
23+
24+
//继承BaseFunction类，重新execute方法
25+
public static class SumFunction extends BaseFunction {
26+
@Override
27+
public void execute(TridentTuple tuple, TridentCollector collector) {
28+
System.out.println("传入进来的内容为：" + tuple);
29+
//获取a、b俩个域
30+
int a = tuple.getInteger(0);
31+
int b = tuple.getInteger(1);
32+
int sum = a + b;
33+
//发射数据
34+
collector.emit(new Values(sum));
35+
}
36+
}
37+
38+
//继承BaseFunction类，重新execute方法
39+
public static class Result extends BaseFunction {
40+
@Override
41+
public void execute(TridentTuple tuple, TridentCollector collector) {
42+
//获取tuple输入内容
43+
System.out.println();
44+
Integer a = tuple.getIntegerByField("a");
45+
Integer b = tuple.getIntegerByField("b");
46+
Integer c = tuple.getIntegerByField("c");
47+
Integer d = tuple.getIntegerByField("d");
48+
System.out.println("a: "+ a + ", b: " + b + ", c: " + c + ", d: " + d);
49+
Integer sum = tuple.getIntegerByField("sum");
50+
System.out.println("sum: "+ sum);
51+
}
52+
}
53+
54+
public static StormTopology buildTopology() {
55+
TridentTopology topology = new TridentTopology();
56+
//设定数据源
57+
FixedBatchSpout spout = new FixedBatchSpout(
58+
new Fields("a", "b", "c", "d"), //声明输入的域字段为"a"、"b"、"c"、"d"
59+
4, //设置批处理大小为1
60+
//设置数据源内容
61+
//测试数据
62+
new Values(1, 4, 7, 10),
63+
new Values(1, 1, 3, 11),
64+
new Values(2, 2, 7, 1),
65+
new Values(2, 5, 7, 2));
66+
//指定是否循环
67+
spout.setCycle(false);
68+
//指定输入源spout
69+
Stream inputStream = topology.newStream("spout", spout);
70+
/**
71+
* 要实现流sqout - bolt的模式 在trident里是使用each来做的
72+
* each方法参数：
73+
* 1.输入数据源参数名称："a", "b", "c", "d"
74+
* 2.需要流转执行的function对象（也就是bolt对象）：new SumFunction()
75+
* 3.指定function对象里的输出参数名称：sum
76+
*/
77+
inputStream.each(new Fields("a", "b", "c", "d"), new SumFunction(), new Fields("sum"))
78+
/**
79+
* 继续使用each调用下一个function（bolt）
80+
* 第一个输入参数为："a", "b", "c", "d", "sum"
81+
* 第二个参数为：new Result() 也就是执行函数，第三个参数为没有输出
82+
*/
83+
.each(new Fields("a", "b", "c", "d", "sum"), new Result(), new Fields());
84+
return topology.build(); //利用这种方式，我们返回一个StormTopology对象，进行提交
85+
}
86+
87+
public static void main(String[] args) throws Exception {
88+
89+
Config conf = new Config();
90+
//设置batch最大处理
91+
conf.setNumWorkers(2);
92+
conf.setMaxSpoutPending(20);
93+
if(args.length == 0) {
94+
LocalCluster cluster = new LocalCluster();
95+
cluster.submitTopology("trident-function", conf, buildTopology());
96+
Thread.sleep(10000);
97+
cluster.shutdown();
98+
} else {
99+
StormSubmitter.submitTopology(args[0], conf, buildTopology());
100+
}
101+
102+
}
103+
}

‎src/com/msj/storm/trident/strategy/StrategyTopology.java‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
package com.msj.storm.trident.strategy;
2+
3+
import backtype.storm.Config;
4+
import backtype.storm.LocalCluster;
5+
import backtype.storm.StormSubmitter;
6+
import backtype.storm.generated.StormTopology;
7+
import backtype.storm.tuple.Fields;
8+
import backtype.storm.tuple.Values;
9+
import storm.trident.Stream;
10+
import storm.trident.TridentTopology;
11+
import storm.trident.testing.FixedBatchSpout;
12+
13+
/**
14+
* @author Vincent.M mengshaojie@188.com
15+
* @date 2017/5/11 下午2:57
16+
* @copyright ©2017 孟少杰 All Rights Reserved
17+
* @desc 测试Trident分组功能
18+
*/
19+
public class StrategyTopology {
20+
21+
public static StormTopology buildTopology() {
22+
TridentTopology topology = new TridentTopology();
23+
//设定数据源
24+
FixedBatchSpout spout = new FixedBatchSpout(
25+
new Fields("sub"), //声明输入的域字段为"sub"
26+
4, //设置批处理大小为4
27+
//设置数据源内容
28+
//测试数据
29+
new Values("java"),
30+
new Values("python"),
31+
new Values("php"),
32+
new Values("c++"),
33+
new Values("ruby"));
34+
//指定是否循环
35+
spout.setCycle(true);
36+
//指定输入源spout
37+
Stream inputStream = topology.newStream("spout", spout);
38+
/**
39+
* 要实现流sqout - bolt的模式 在trident里是使用each来做的
40+
* each方法参数：
41+
* 1.输入数据源参数名称："sub"
42+
* 2.需要流转执行的function对象（也就是bolt对象）：new WriteFunction()
43+
* 3.指定function对象里的输出参数名称,没有则不输出任何内容
44+
*/
45+
inputStream
46+
//随机分组：shuffle
47+
.shuffle()
48+
//分区分组：partitionBy
49+
//.partitionBy(new Fields("sub"))
50+
//全局分组：global
51+
//.global()
52+
//广播分组：broadcast
53+
//.broadcast()
54+
.each(new Fields("sub"), new WriteFunction(), new Fields()).parallelismHint(4);
55+
return topology.build(); //利用这种方式，我们返回一个StormTopology对象，进行提交
56+
}
57+
58+
public static void main(String[] args) throws Exception {
59+
60+
Config conf = new Config();
61+
//设置batch最大处理
62+
conf.setNumWorkers(2);
63+
conf.setMaxSpoutPending(20);
64+
if(args.length == 0) {
65+
LocalCluster cluster = new LocalCluster();
66+
cluster.submitTopology("trident-strategy", conf, buildTopology());
67+
Thread.sleep(5000);
68+
cluster.shutdown();
69+
} else {
70+
StormSubmitter.submitTopology(args[0], conf, buildTopology());
71+
}
72+
73+
}
74+
}

‎src/com/msj/storm/trident/strategy/WriteFunction.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package com.msj.storm.trident.strategy;
2+
3+
import org.apache.commons.logging.Log;
4+
import org.apache.commons.logging.LogFactory;
5+
import storm.trident.operation.BaseFunction;
6+
import storm.trident.operation.TridentCollector;
7+
import storm.trident.tuple.TridentTuple;
8+
9+
import java.io.FileWriter;
10+
11+
/**
12+
* @author Vincent.M mengshaojie@188.com
13+
* @date 2017/5/11 下午2:57
14+
* @copyright ©2017 孟少杰 All Rights Reserved
15+
* @desc 测试Trident分组功能
16+
*/
17+
public class WriteFunction extends BaseFunction {
18+
/** serialVersionUID */
19+
private static final long serialVersionUID = 1L;
20+
21+
private FileWriter writer ;
22+
23+
private static final Log log = LogFactory.getLog(WriteFunction.class);
24+
25+
@Override
26+
public void execute(TridentTuple tuple, TridentCollector collector) {
27+
String text = tuple.getStringByField("sub");
28+
try {
29+
if(writer == null){
30+
if(System.getProperty("os.name").equals("Windows 10")){
31+
writer = new FileWriter("D:\\099_test\\" + this);
32+
} else if(System.getProperty("os.name").equals("Windows 8.1")){
33+
writer = new FileWriter("D:\\099_test\\" + this);
34+
} else if(System.getProperty("os.name").equals("Windows 7")){
35+
writer = new FileWriter("D:\\099_test\\" + this);
36+
} else if(System.getProperty("os.name").equals("Linux")){
37+
System.out.println("----:" + System.getProperty("os.name"));
38+
writer = new FileWriter("/usr/local/temp/" + this);
39+
}
40+
}
41+
log.info("【write】： 写入文件");
42+
writer.write(text);
43+
writer.write("\n");
44+
writer.flush();
45+
46+
} catch (Exception e) {
47+
e.printStackTrace();
48+
}
49+
}
50+
}

‎src/com/msj/storm/trident/wordcount/ResultFunction.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
package com.msj.storm.trident.wordcount;
2+
3+
import storm.trident.operation.BaseFunction;
4+
import storm.trident.operation.TridentCollector;
5+
import storm.trident.tuple.TridentTuple;
6+
/**
7+
* @author Vincent.M mengshaojie@188.com
8+
* @date 2017/5/11 下午2:57
9+
* @copyright ©2017 孟少杰 All Rights Reserved
10+
* @desc 测试Trident WordCount
11+
*/
12+
public class ResultFunction extends BaseFunction {
13+
/** serialVersionUID */
14+
private static final long serialVersionUID = 1L;
15+
16+
@Override
17+
public void execute(TridentTuple tuple, TridentCollector collector) {
18+
//获取tuple输入内容
19+
String sub = tuple.getStringByField("sub");
20+
Long count = tuple.getLongByField("count");
21+
System.out.println(sub +" : "+ count);
22+
}
23+
}

‎src/com/msj/storm/trident/wordcount/SplitFunction.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
package com.msj.storm.trident.wordcount;
2+
3+
import storm.trident.operation.BaseFunction;
4+
import storm.trident.operation.TridentCollector;
5+
import storm.trident.tuple.TridentTuple;
6+
import backtype.storm.tuple.Values;
7+
/**
8+
* @author Vincent.M mengshaojie@188.com
9+
* @date 2017/5/11 下午2:57
10+
* @copyright ©2017 孟少杰 All Rights Reserved
11+
* @desc 测试Trident WordCount
12+
*/
13+
public class SplitFunction extends BaseFunction {
14+
/** serialVersionUID */
15+
private static final long serialVersionUID = 1L;
16+
17+
@Override
18+
public void execute(TridentTuple tuple, TridentCollector collector) {
19+
String subjects = tuple.getStringByField("subjects");
20+
//获取tuple输入内容
21+
//逻辑处理，然后发射给下一个组件
22+
for(String sub : subjects.split(" ")) {
23+
collector.emit(new Values(sub));
24+
}
25+
}
26+
}

0 commit comments

Comments
 (0)