Storm Trident API

　　在Storm Trident中有五种操作类型

　　Apply Locally:本地操作，所有操作应用在本地节点数据上，不会产生网络传输
　　Repartitioning:数据流重定向，单纯的改变数据流向，不会改变数据内容，这部分会有网络传输
　　Aggragation:聚合操作，会有网络传输
　　Grouped streams上的操作
　　Merge和Join

一Apply Locally

　　1.functions函数操作

　　函数的作用是接收一个tuple(需指定接收tuple的哪个字段)，输出0个或多个tuples。输出的新字段值会被追加到原始输入tuple的后面，如果一个function不输出tuple，那就意味这这个tuple被过滤掉了，例如下面的例子：

1 class AddAndSubFuction extends BaseFunction{
2 
3         public void execute(TridentTuple tuple, TridentCollector collector) {
4             int res1 = tuple.getInteger(0);
5             int res2 = tuple.getInteger(1);
6             int sub = res1 > res2 ? res1 - res2 : res2 - res1;
7             collector.emit(new Values(res1+res2,sub));
8         }
9     }

　　2.Filter过滤操作

　　Filters很简单，接收一个tuple，并决定是否保留这个tuple，例如

1 class ScoreFilter extends BaseFilter{
2 
3         public boolean isKeep(TridentTuple tuple) {
4             return tuple.getInteger(0) >= 60;
5         }
6     }

　　上述Filter过滤调成绩小于60的tuple.

　　3.partitionAggregate

　　PartitionAggregate的作用对每个Partition中的tuple进行聚合，与前面的函数在原tuple后面追加数据不同，PartitionAggregate的输出会直接替换掉输入的tuple，仅数据PartitionAggregate中发射的tuple。

　　TridentAPI提供了三个聚合器接口：CombinerAggregator，ReducerAggregator，Aggregator

　　我们先来看一看CombinerAggregator，CombinerAggregator接口的定义如下：

public interface CombinerAggregator<T> extends Serializable {
    T init(TridentTuple tuple);
    T combine(T val1, T val2);
    T zero();
}

　　CombinerAggregator接口只返回一个tuple，并且这个tuple也只包含一个field。init方法会先执行，它负责预处理每一个接收到的tuple，然后再执行combine函数来计算收到的tuples直到最后一个tuple到达，当所有tuple处理完时，CombinerAggregator会发射zero函数的输出，比如CombinerAggregator的实现类Count的定义如下：

public class Count implements CombinerAggregator<Long> {

    @Override
    public Long init(TridentTuple tuple) {
        return 1L;
    }

    @Override
    public Long combine(Long val1, Long val2) {
        return val1 + val2;
    }

    @Override
    public Long zero() {
        return 0L;
    }
    
}

　　当你使用aggregate 方法代替PartitionAggregate时，CombinerAggregator的好处就体现出来了，因为Trident会自动优化计算，在网络传输tuples之前做局部聚合。

　　我们再来看一下ReducerAggregator，ReducerAggregator的定义如下：

public interface ReducerAggregator<T> extends Serializable {
    T init();
    T reduce(T curr, TridentTuple tuple);
}

　　ReducerAggregator通过init方法提供一个初始值，然后为输入的每个tuple迭代这个值，最终产生一个唯一的tuple并输出，定义一个实例如下：

 public class ReducerCount implements ReducerAggregator<Long>{
     @Override
     public Long init() {
           return 0L;
     }
     @Override
     public Long reduce(Long curr, TridentTuple tuple) {
           return curr + 1;
     }
 }

　　最后看一下通用的聚合器Aggregator，它的定义如下：

public interface Aggregator<T> extends Operation {
    T init(Object batchId, TridentCollector collector);
    void aggregate(T val, TridentTuple tuple, TridentCollector collector);
    void complete(T val, TridentCollector collector);
}

　　Aggregator接口可以发射含任意数量属性的任意数据量的tuples,并且可以在执行过程中的任何时候发射:
　　init:在处理数据之前被调用，它的返回值会作为一个状态值传递给aggregate和complete方法
　　aggregate：用来处理每一个输入的tuple，它可以更新状态值也可以发射tuple
　　complete：当所有tuple都被处理完成后被调用

　　有时候我们需要执行多个聚合器，这在Trident中称为chaining

　　4.projection投影操作

　　投影操作的作用是仅仅保留stream指定字段的数据，和关系数据库中投影的概念类似

二Repartitioning重定向操作

　　重定向操作是如何在各个任务间对tuples进行分区。分区的数量也有可能改变重定向的结果。重定向需要网络传输，下面介绍下重定向函数：

shuffle：通过随机分配算法来均衡tuple到各个分区
broadcast：每个tuple都被广播到所有的分区，这种方式在drcp时非常有用，比如在每个分区上做stateQuery
partitionBy：根据指定的字段列表进行划分，具体做法是用指定字段列表的hash值对分区个数做取模运算，确保相同字段列表的数据被划分到同一个分区
global：所有的tuple都被发送到一个分区，这个分区用来处理整个Stream
batchGlobal：一个Batch中的所有tuple都被发送到同一个分区，不同的Batch会去往不同的分区
Partition：通过一个自定义的分区函数来进行分区，这个自定义函数实现了 backtype.storm.grouping.CustomStreamGrouping

三Aggragation聚合操作

　　Trident有aggregate和 persistentAggregate方法来做聚合操作。aggregate是独立的运行在Stream的每个Batch上的，而persistentAggregate则是运行在Stream的所有Batch上并把运算结果存储在state source中。运行aggregate方法做全局聚合。当你用到 ReducerAggregator或Aggregator时，Stream首先被重定向到一个分区中，然后其中的聚合函数便在这个分区上运行。当你用到CombinerAggregator时，Trident会首先在每个分区上做局部聚合，然后把局部聚合后的结果重定向到一个分区，因此使用CombinerAggregator会更高效，可能的话我们需要优先考虑使用它。

四Grouped streams

　　GroupBy操作是根据特定的字段对流进行重定向的，还有，在一个分区内部，每个相同字段的tuple也会被Group到一起。如果你在grouped Stream上面运行aggregators，聚合操作会运行在每个Group中而不是整个Batch。persistentAggregate也能运行在GroupedSteam上，不过结果会被保存在MapState中，其中的key便是分组的字段。当然，aggregators在GroupedStreams上也可以串联。

五Merge和Join

　　api的最后一部分便是如何把各种流汇聚到一起。最简单的方式就是把这些流汇聚成一个流。我们可以这么做：　　

topology.merge(stream1, stream2, stream3);

　　另一种合并流的方式就是join。一个标准的join就像是一个sql,必须有标准的输入，因此，join只针对符合条件的Stream。join应用在来自Spout的每一个小Batch中。join时候的tuple会包含：
　　1.join的字段，如Stream1中的key和Stream2中的x

　　2.所有非join的字段，根据传入join方法的顺序，a和b分别代表steam1的val1和val2，c代表Stream2的val1

　　当join的是来源于不同Spout的stream时，这些Spout在发射数据时需要同步，一个Batch所包含的tuple会来自各个Spout。