【Storm】Trident API和概念(Operation类)
作者：互联网
一、Trident Spout
ITridentSpout：是最通用的spout，可以支持事务或者不透明事务定义；
IBatchSpout：一个非事务spout；
IPartitionedTridentSpout：分区事务spout，从数据源（kafka集群）读分区数据；
IOpaquePartitionedTridentSpout：不透明分区事务spout，从数据源读分区数据；
1、ITridentSpout
（1）ITridentSpout<T> 接口类，有2个内部接口类BatchCoordinator和Emitter，4个方法
（2）BatchCoordinator<X> 接口类，有4个方法
（3）Emitter<X>接口类，有3个方法
2、IBatchSpout
IBatchSpout接口，FixedBatchSpout类实现了IBatchSpout接口。IBatchSpout有6个方法；
3、IPartitionedTridentSpout
（1）IPartitionedTridentSpout<Partitions,Partition extends ISpoutPartition,T>，拥有2个类和4个方法
（2）IPartitionedTridentSpout.Coordinator<Partitions>，Partitions是自定义的类；getPartitionsForBatch是返回当前数据源的所有分区；
（3）IPartitionedTridentSpout.Emitter<Partitions,Partition extends ISpoutPartition,X>，入参是partitions集合，自定义的partition要继承ISpoutPartition类；getOrderedPartitions获得下一个分区；refreshPartitions是如果有新增分区，刷新分区；
4、IOpaquePartitionedTridentSpout
（1）IOpaquePartitionedTridentSpout<Partitions,Partition extends ISpoutPartition,M>
（2）IOpaquePartitionedTridentSpout.Coordinator<Partitions>
（3）IOpaquePartitionedTridentSpout.Emitter<Partitions,Partition extends ISpoutPartition,M>
二、Trident Bolt
唯一显性Bolt接口：ITridentBatchBolt，但很少用；
Trident编程的特点就是Stream
Trident的topology会被编译成尽可能搞笑的Storm topology，只有在需要对数据进行repartition的时候（如groupby或者shuffle）才会把tuple通过network发送出去，如果你有一个trident如下：
被编译成storm topology如下
三、Trident概念之Operation
所有的类和接口在 storm.trident.operation 包
Trident API可以分为Spout操作和Bolt操作，对于Bolt操作提供常见的流数据分析操作。
Bolt Trident提供了五种类型的操作：
- Apply Locally: 本地分区操作(Partition-local operations)，所有操作应用在本地节点数据上，不会产生网络传输
- Repartitioning: 重分区操作，数据流重定向，单纯的改变数据流向，不会改变数据内容，这部分会有网络传输
- Aggragation: 聚合操作，会有网络传输
- Grouped streams: 流分组操作操作
- 合并(meger)和连接(join)操作
小结：上面提到了Trident实际上是通过把函数应用到每个节点的Batch上的数据以实现并行，而应用的这些函数就是TridentAPI，下面我们就具体介绍一下TridentAPI的各种操作。
Trident五种操作详解
3.1、Apply Locally本地操作：操作都应用在本地节点的Batch上，不会产生网络传输
3.1.1 Functions: 函数操作
函数的作用是接收一个tuple(需指定接收tuple的哪个字段)，输出0个或多个tuples。输出的新字段值会被追加到原始输入tuple的后面，如果一个function不输出tuple，那就意味这这个tuple被过滤掉了，下面举例说明：
定义一个Function：
public class MyFunction extends BaseFunction{
private static final long serialVersionUID = 1L;
/**
* 在每个元组上面执行该逻辑函数，并且发射0个或多个元组
*
* @param tuple 传入的元组
* @param collector 用于发射元组的收集器实例
*/
@Override
public void execute(TridentTuple tuple, TridentCollector collector) {
// 接收第一个field
for (int i = 0; i < tuple.getInteger(0); i++) {
collector.emit(new Values(i));
}
}
}
小结：Function实际上就是对经过Function函的tuple做一些操作以改变其内容。
比如我们处理一个“mystream”的数据流，它有三个字段分别是[“a”, “b”, “c”] ，数据流中tuple的内容是：
[1, 2, 3]
[4, 1, 6]
[3, 0, 8]
我们将每个元组都经过以下MyFunction操作：
//将每个tuple的字段"b"应用于MyFunction，并且产生新字段"d"追加到原tuple的字段中
mystream.each(new Fields("b"),new MyFunction(),new Fields('d'))
它意思是接收输入的每个tuple “b”字段得值，把函数结算结果做为新字段“d”追加到每个tuple后面，然后发射出去。
最终运行结果会是每个tuple有四个字段[“a”, “b”, “c”, “d”]，每个tuple的内容变成了：
//[1,2,3]的emit，其中0和1是新字段“d”
[1, 2, 3, 0]
[1, 2, 3, 1]
//[4,1,6]的emit，其中0是新字段“d”
[4, 1, 6, 0]
//[3,0,8]，b列是0，不满足需求，过滤掉了
小结：我们注意到，如果一个function发射多个tuple时，每个发射的新tuple中仍会保留原来老tuple的数据。
3.1.2 Filters:过滤操作
Filters很简单，接收一个tuple并决定是否保留这个tuple。举个例子，定义一个Filter：
public class MyFilter extends BaseFilter{
@Override
public boolean isKeep(TridentTuple tuple) {
return tuple.getInteger(0) == 1 && tuple.getInteger(1) == 2;
}
}
假设我们的tuples有这个几个字段 [“a”, “b”, “c”]:
[1, 2, 3]
[2, 1, 1]
[2, 3, 4]
然后运行我们的Filter：
mystream.each(new Fields("b", "a"), new MyFilter());
则最终得到的tuple是:
[2, 1, 1]
说明第一个和第三个不满足条件，都被过滤掉了。
小结：Filter就是一个过滤器，它决定是否需要保留当前tuple。
3.1.3 PartitionAggregate
PartitionAggregate的作用对每个Partition中的tuple进行聚合，与前面的函数在原tuple后面追加数据不同，PartitionAggregate的输出会直接替换掉输入的tuple，仅数据PartitionAggregate中发射的tuple。下面举例说明：
定义一个累加的PartitionAggregate：
mystream.partitionAggregate(new Fields("b"), new Sum(), new Fields("sum"))；
假设我们的Stream包含两个字段 [“a”, “b”]，各个Partition的tuple内容是：
Partition 0: [“a”, 1] [“b”, 2]
Partition 1: [“a”, 3] [“c”, 8]
Partition 2: [“e”, 1] [“d”, 9] [“d”, 10]
输出的内容只有一个字段“sum”,值是：
Partition 0: [3]
Partition 1: [11]
Partition 2: [20]
TridentAPI提供了三个聚合器的接口：CombinerAggregator, ReducerAggregator, Aggregator.
CombinerAggregator操作
CombinerAggregator只返回单个tuple，并且这个tuple只包含一个Field。每个元组首先都经过init函数进行预处理，然后在执行combine函数来计算接受到的tuple，直到最后一个tuple到达。如果分区内没有tuple，则会通过zero函数发射结果。
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实现来计数：
public class CombinerCount implements CombinerAggregator<Integer>{
@Override
public Integer init(TridentTuple tuple) {
return 1;
}
@Override
public Integer combine(Integer val1, Integer val2) {
return val1 + val2;
}
@Override
public Integer zero() {
return 0;
}
}
小结：当你使用aggregate 方法代替PartitionAggregate时，CombinerAggregator的好处就体现出来了，因为Trident会自动优化计算，在网络传输tuples之前做局部聚合。
ReducerAggregator操作
ReducerAggregator通过init方法提供一个初始值，然后每个输入的tuple迭代这个值，最后产生一个唯一的tuple输出。
public interface ReducerAggregator<T> extends Serializable {
T init();
T reduce(T curr, TridentTuple tuple);
}
比如同样使用RecuerAggregator来实现计数器：
public class ReducerCount implements ReducerAggregator<Long>{
@Override
public Long init() {
return 0L;
}
@Override
public Long reduce(Long curr, TridentTuple tuple) {
return curr + 1;
}
}
Aggregator操作
执行聚合操作最通用的接口就是Aggregator了，它能够发射任意数量的元组，每个元组可以包含任意数量的字段。
public interface Aggregator<T> extends Operation {
T init(Object batchId, TridentCollector collector);
void aggregate(T state, TridentTuple tuple, TridentCollector collector);
void complete(T state, TridentCollector collector);
}
它的执行流程是：
- 在处理Batch之前调用init方法，它返回一个聚合的状态值，传递给aggregate和complete方法。
- 为批处理分区中的每个tuple调用aggregate方法，此方法可以更新状态值，也可以发射元组。
- 当aggregator处理完Batch分区的所有元组后调用complete方法。
使用Aggregator来实现计数器：
public class CountAgg extends BaseAggregator<CountState>{
static class CountState {
long count = 0;
}
@Override
public CountState init(Object batchId, TridentCollector collector) {
return new CountState();
}
@Override
public void aggregate(CountState val, TridentTuple tuple, TridentCollector collector) {
val. count+=1;
}
@Override
public void complete(CountState val, TridentCollector collector) {
collector.emit( new Values(val. count));
}
}
projection 投影操作
projection操作用于只保留指定的字段，比如元组有字段["a","b","c","d"]，通过以下投影操作，输出流只会包含["c","d"]。
mystream.projection(new Fields("c","d"));
3.2 Repartition 重分区操作
Repartition操作运行一个函数来改变元组在任务之间的分布，调整分区数也可能会导致Repartition操作。重分区操作会引发网络传输。下面是重分区的相关函数：
- shuffle：使用随机算法来均衡tuple到每个分区。
- broadcast：每个tuple被广播到所有分区上，使用DRPC时使用这种方法比较多，比如每个分区上做stateQuery。
- global：所有tuple都发送到一个分区上，这个分区用来处理stream。
- batchGlobal：一个batch中的所有tuple会发送到一个分区中，不同batch的元组会被发送到不同分区上。
- partition：通过一个自定义的分区函数来进行分区，这个自定义函数需要实现
org.apache.storm.grouping.CustomStreamGrouping。
3.3 Aggragation 聚合操作
Trident提供了aggregate和persistentAggregate方法，aggregate运行在每个batch中，而persistentAggregate将聚合所有Batch，并将结果保存在一个状态源上。
我们前面讲的aggregate、CombinerAggregator和ReducerAggregator运行在patitionAggregation上是本地分区操作。如果直接作用于流上，则是对全局进行聚合。
在对全局流进行聚合时，Aggregator和ReducerAggregator会首先重分区到一个单分区，然后在该分区上执行聚合函数。而CombinerAggregator则会首先聚合每个分区，然后重分区到单个分区，在网络传输中完成聚合操作。所以我们应该尽量用CombinerAggregator，因为它更加高效。
mystream.aggregate(new Count(),new Fields("count"));
3.4 grouped streams 流分组操作
GroupBy操作是根据特定的字段对流进行重定向的，还有，在一个分区内部，每个相同字段的tuple也会被Group到一起，下面这幅图描述了这个场景：
如果你在grouped Stream上面运行aggregators，聚合操作会运行在每个Group中而不是整个Batch。persistentAggregate也能运行在GroupedSteam上，不过结果会被保存在MapState中，其中的key便是分组的字段。
当然，aggregators在GroupedStreams上也可以串联。
3.5 合并(meger)和连接(join)操作
Trident可以允许我们将不同流组合在一起，通过TridentTopology.merge()方法操作。
//合并流会以第一个流的输出字段来命名
topology.mege(stream1,stream2,stream3);
另一种合并流的方式是连接，类似于SQL那样的连接，要求输入是有限的。所以Trident的join只适用于来自Spout的每个小Bath之间。
比如有一个流包含["key1","val1","val2"]，另一个流包含["key2","val1","val2"]，通过以下连接操作：
//Trident需要join之后的流重新命名，因为输入流可能存在重复 字段。
mystream.join(stream1,new Fields("key1"),stream2,new Fields("key2"),new Fields("key","a","b","c"，"d"))
标签：tuple,分区,Trident,API,new,操作,Operation,public 来源： https://blog.csdn.net/u014028317/article/details/100712816