Trident学习笔记（一）

1. Trident入门

Trident

-------------------

　三叉戟

　storm高级抽象，支持有状态流处理；

　好处是确保消费被处理一次；

　以小批次方式处理输入流，得到精准一次性处理；

　不再使用bolt，使用functions、aggreates、filters以及states。

　Trident Tuple: trident top的数据模型，trident处理数据的单元；

　　　　　　　　每个tuple有预定义的字段列表构成，字段类型可以是byte；

　　　　　　　　character，integer，long，float，double，Boolean or byte array。

　Trident functions: 包含修改tuple的业务逻辑，输入的是tuple的字段，输出多个tuple。

1import org.apache.storm.trident.operation.BaseFunction;
2import org.apache.storm.trident.operation.TridentCollector;
3import org.apache.storm.trident.tuple.TridentTuple;
4import org.apache.storm.tuple.Values;
5
6/**
7 * 求和函数
8 */
9public class SumFunction extends BaseFunction { @Override public void execute(TridentTuple input, TridentCollector collector) { Integer num1 = input.getInteger(0); Integer num2 = input.getInteger(1); int sum = num1 + num2; collector.emit(new Values(sum)); } }

如果tuple有a, b, c, d四个field，只有a和b作为输入传给function，functions会生成新的sum字段，

sum字段和输入的元祖进行合并，生成一个完成tuple，因此，新的tuple的总和字段个数是a, b, c, d, sum。

Trident Filter

--------------------

　　1. 描述

　　获取字段集合作为输入，输出boolean，如果反悔true，tuple在流中保留，否则删除，

　　a, b, c, d, sum是元祖的字段，sum作为输入传递给filter，判断sum是否为偶数，

　　如果是偶数，tuple(a, b, c, d, sum)保留，否则tuple删除。

　　2. 代码

1import org.apache.storm.trident.operation.BaseFilter;
2import org.apache.storm.trident.tuple.TridentTuple;
3
4/**
5 * 校验是否是偶数的过滤器
6 */
7public class CheckEvenFilter extends BaseFilter {
8
9    @Override
10    public boolean isKeep(TridentTuple input) {
11        Integer sum = input.getInteger(0); if (sum % 2 == 0) { return true; } return false; } }

Trident projections

--------------------

　　1. 描述

　　　投影操作中，trident值保留在投影中制定的字段，

　　　x, y, z --> projection(x) --> x

　　2. 调用投影的方式

　　　mystream.project(new fields("x"));

写一个topology

1import org.apache.storm.trident.operation.BaseFunction;
2import org.apache.storm.trident.operation.TridentCollector;
3import org.apache.storm.trident.tuple.TridentTuple;
4
5public class PrintFunction extends BaseFunction {
6
7    @Override
8    public void execute(TridentTuple input, TridentCollector collector) {
9        Integer sum = input.getInteger(0);
10        System.out.println(this.getCLass.getSimpleName + ": " + sum);
11    }
12    
13}
14
15import com.google.common.collect.ImmutableList;
16import org.apache.storm.Config;
17import org.apache.storm.LocalCluster;
18import org.apache.storm.trident.Stream;
19import org.apache.storm.trident.TridentTopology;
20import org.apache.storm.trident.testing.FeederBatchSpout;
21import org.apache.storm.tuple.Fields;
22import org.apache.storm.tuple.Values;
23
24public class TridentTopologyApp {
25
26    public static void main(String[] args) {
27        // 创建topology
28        TridentTopology topology = new TridentTopology();
29
30        // 创建spout
31        FeederBatchSpout testSpout = new FeederBatchSpout(ImmutableList.of("a", "b", "c", "d"));
32
33        // 创建流
34        Stream stream = topology.newStream("spout", testSpout);
35        stream.shuffle().each(new Fields("a", "b"), new SumFunction(), new Fields("sum")).parallelismHint(1)
36                .shuffle().each(new Fields("sum"), new CheckEvenFilter()).parallelismHint(1)
37                .shuffle().each(new Fields("sum"), new PrintFunction(), new Fields("xxx")).parallelismHint(1);
38
39        // 本地提交
40        LocalCluster cluster = new LocalCluster();
41        cluster.submitTopology("TridentDemo", new Config(), topology.build());
42
43        // 测试数据
44        testSpout.feed(ImmutableList.of(new Values(1, 2, 3, 4)));
45        testSpout.feed(ImmutableList.of(new Values(2, 3, 4, 5)));
46        testSpout.feed(ImmutableList.of(new Values(3, 4, 5, 6)));
47        testSpout.feed(ImmutableList.of(new Values(4, 5, 6, 7)));
48    }
49
50}

输出结果

1SumFunction:1, 2
2CheckEvenFilter:3
3PrintFunction: 3
4SumFunction:2, 3
5CheckEvenFilter:5
6PrintFunction: 5
7SumFunction:3, 4
8CheckEvenFilter:7
9PrintFunction: 7
10SumFunction:4, 5
11CheckEvenFilter:9
12PrintFunction: 9

加入一个求平均数的函数

1import org.apache.storm.trident.operation.BaseFunction;
2import org.apache.storm.trident.operation.TridentCollector;
3import org.apache.storm.trident.tuple.TridentTuple;
4
5/**
6 * 求平均值方法
7 */
8public class AverageFunction extends BaseFunction {
9
10    @Override
11    public void execute(TridentTuple input, TridentCollector collector) {
12        int a = input.getIntegerByField("a");
13        int b = input.getIntegerByField("b");
14        int c = input.getIntegerByField("c");
15        int d = input.getIntegerByField("d");
16        int sum = input.getIntegerByField("sum");
17        float avg = (float) ((a+b+c+d+sum) / 5.0);
18        System.out.println(this.getClass().getSimpleName() + ": avg = " + avg);
19    }
20
21}
22
23import com.google.common.collect.ImmutableList;
24import org.apache.storm.Config;
25import org.apache.storm.LocalCluster;
26import org.apache.storm.trident.Stream;
27import org.apache.storm.trident.TridentTopology;
28import org.apache.storm.trident.testing.FeederBatchSpout;
29import org.apache.storm.tuple.Fields;
30import org.apache.storm.tuple.Values;
31
32public class TridentTopologyApp {
33
34    public static void main(String[] args) {
35        // 创建topology
36        TridentTopology topology = new TridentTopology();
37
38        // 创建spout
39        FeederBatchSpout testSpout = new FeederBatchSpout(ImmutableList.of("a", "b", "c", "d"));
40
41        // 创建流
42        Stream stream = topology.newStream("spout", testSpout);
43        stream.shuffle().each(new Fields("a", "b"), new SumFunction(), new Fields("sum")).parallelismHint(1)
44                .shuffle().each(new Fields("sum"), new CheckEvenFilter()).parallelismHint(1)
45                .shuffle().each(new Fields("sum"), new PrintFunction(), new Fields("res")).parallelismHint(1)
46                .shuffle().each(new Fields("a", "b", "c", "d", "sum"), new AverageFunction(), new Fields("avg")).parallelismHint(1);
47
48        // 本地提交
49        LocalCluster cluster = new LocalCluster();
50        cluster.submitTopology("TridentDemo", new Config(), topology.build());
51
52        // 测试数据
53        testSpout.feed(ImmutableList.of(new Values(1, 2, 3, 4)));
54        testSpout.feed(ImmutableList.of(new Values(2, 3, 4, 5)));
55        testSpout.feed(ImmutableList.of(new Values(3, 4, 5, 6)));
56        testSpout.feed(ImmutableList.of(new Values(4, 5, 6, 7)));
57    }
58
59}

2. Trident聚合函数

分区聚合

1import com.google.common.collect.ImmutableList;
2import org.apache.storm.Config;
3import org.apache.storm.LocalCluster;
4import org.apache.storm.trident.Stream;
5import org.apache.storm.trident.TridentTopology;
6import org.apache.storm.trident.testing.FeederBatchSpout;
7import org.apache.storm.tuple.Fields;
8import org.apache.storm.tuple.Values;
9
10public class TridentTopologyApp2 {
11
12    public static void main(String[] args) {
13        // 创建topology
14        TridentTopology topology = new TridentTopology();
15
16        // 创建spout
17        FeederBatchSpout testSpout = new FeederBatchSpout(ImmutableList.of("a", "b"));
18
19        // 创建流
20        Stream stream = topology.newStream("testSpout", testSpout);
21        stream.shuffle().each(new Fields("a", "b"), new MyFilter1()).parallelismHint(1)
22                .global().each(new Fields("a", "b"), new MyFilter2()).parallelismHint(1)
23                .partitionBy(new Fields("a"))
24                //.each(new Fields("a", "b"), new MyFunction1(), new Fields("none")).parallelismHint(1)
25                .partitionAggregate(new Fields("a"), new MyCount(), new Fields("count"))
26                .each(new Fields("count"), new MyPrintFunction1(), new Fields("xxx")).parallelismHint(1);
27
28        // 本地提交
29        LocalCluster cluster = new LocalCluster();
30        cluster.submitTopology("TridentDemo2", new Config(), topology.build());
31
32        // 测试数据
33        testSpout.feed(ImmutableList.of(new Values(1, 2)));
34        testSpout.feed(ImmutableList.of(new Values(2, 3)));
35        testSpout.feed(ImmutableList.of(new Values(2, 4)));
36        testSpout.feed(ImmutableList.of(new Values(3, 5)));
37    }
38
39}

批次聚合

Trident学习笔记（一）

1. Trident入门

2. Trident聚合函数

3. 自定义聚合函数-Sum-SumAsAggregator

评论区

相关推荐