Storm Trident API 使用详解

小咪咪 2022-06-02 23:25 250阅读 0赞

一、概述

Storm Trident中的核心数据模型就是“Stream”，也就是说，Storm Trident处理的是Stream，但是实际上Stream是被成批处理的，Stream被切分成一个个的Batch分布到集群中，所有应用在Stream上的函数最终会应用到每个节点的Batch中，实现并行计算，具体如下图所示：

![SouthEast][]

在Trident中有五种操作类型：

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

小结：上面提到了Trident实际上是通过把函数应用到每个节点的Batch上的数据以实现并行，而应用的这些函数就是TridentAPI，下面我们就具体介绍一下TridentAPI的各种操作。

二、Trident五种操作详解

2.1 Apply Locally本地操作：操作都应用在本地节点的Batch上，不会产生网络传输

2.1.1 Functions:函数操作

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

*  定义一个Function：

public class MyFunction extends BaseFunction {
          @Override
          public void execute(TridentTuple tuple, TridentCollector collector) {
               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\]

*  我们运行我们的Function：

java mystream.each(new Fields("b"), new MyFunction(), new Fields("d")))；

它意思是接收输入的每个tuple “b”字段得值，把函数结算结果做为新字段“d”追加到每个tuple后面，然后发射出去。

*  最终运行结果会是每个tuple有四个字段\[“a”, “b”, “c”, “d”\]，每个tuple的内容变成了：

\[1, 2, 3, 0\] \[1, 2, 3, 1\] \[4, 1, 6, 0\]

小结：我们注意到，如果一个function发射多个tuple时，每个发射的新tuple中仍会保留原来老tuple的数据。

2.1.2 Filters:过滤操作

*  Filters很简单，接收一个tuple并决定是否保留这个tuple。举个例子，定义一个Filter：

public class MyFilter extends BaseFilter {
         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：

java mystream.each(new Fields("b", "a"), new MyFilter())；

*  则最终得到的tuple是 ：

\[2, 1, 1\]

说明第一个和第三个不满足条件，都被过滤掉了。

小结：Filter就是一个过滤器，它决定是否需要保留当前tuple。

2.1.3 PartitionAggregate

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

*  定义一个累加的PartitionAggregate：

java 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, and Aggregator.

我们先看一下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：

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

ReducerAggregator通过init方法提供一个初始值，然后为每个输入的tuple迭代这个值，最后生产处一个唯一的tuple输出，下面举例说明：

*  定义一个ReducerAggregator接口实现技术器的例子：

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,并且可以在执行过程中的任何时候发射:

1.  init:在处理数据之前被调用，它的返回值会作为一个状态值传递给aggregate和complete方法
2.  aggregate：用来处理每一个输入的tuple，它可以更新状态值也可以发射tuple
3.  complete：当所有tuple都被处理完成后被调用

下面举例说明：

*  定义一个实现来完成一个计数器：

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));
         }
      }

有时候我们需要同时执行多个聚合器，这在Trident中被称作chaining,使用方法如下：

java mystream.chainedAgg() .partitionAggregate(new Count(), new Fields("count")) .partitionAggregate(new Fields("b"), new Sum(), new Fields("sum")) .chainEnd();

这点代码会在每个Partition上运行count和sum函数，最终输出一个tuple：[“count”, “sum”]

projection:投影操作

投影操作作用是仅保留Stream指定字段的数据，比如有一个Stream包含如下字段： [“a”, “b”, “c”, “d”]

运行如下代码：

java mystream.project(new Fields("b", "d"))

则输出的流仅包含 [“b”, “d”]字段。

2.2 Repartitioning重定向操作

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

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

2.3 Aggragation聚合操作

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

java mystream.aggregate(new Count(), new Fields("count"))；

和paritionAggregate一样，aggregators的聚合也可以串联起来，但是如果你把一个 CombinerAggregator和一个非CombinerAggregator串联在一起，Trident是无法完成局部聚合优化的。

2.4 grouped streams

GroupBy操作是根据特定的字段对流进行重定向的，还有，在一个分区内部，每个相同字段的tuple也会被Group到一起，下面这幅图描述了这个场景：

如果你在grouped Stream上面运行aggregators，聚合操作会运行在每个Group中而不是整个Batch。persistentAggregate也能运行在GroupedSteam上，不过结果会被保存在MapState中，其中的key便是分组的字段。
         当然，aggregators在GroupedStreams上也可以串联。

2.5 Merge和Joins:

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

java 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。

`Trident`是Storm中最为核心的概念，在做Strom开发的过程中，绝大部分情况下我们都会使用Trident，而不是使用传统的Spout、Bolt。Trident是Storm原语的高级封装，学会Trident之后，将会使得我们Storm开发变得非常简单。

一、什么是Storm Trident ？

简而言之：Trident是编写Storm Topology的一套高级框架，是对传统Spout、Bolt的高级封装。在学习Trident之前，我们都是都Spout、Bolt的相关API来编写一个Topology，在学习了Trident之后，我们会使用Trident API来编写Topology。

可以将StormTopology与`TridentTopology`的关系，类比为JDBC与ORM框架(mybatis、hibernate)之间的关系，后者是前者的高级封装，功能相同，但是可以极大的减少我们的开发的工作量。

当然，就像我们学习JDBC与ORM框架一样，JDBC可能很容易理解，但是学习ORM框架，可能就相对复杂一点。甚至学习ORM框架的时间可能要比学习JDBC的时间要更长，但是一旦我们学会了ORM框架，可能就再也不想去使用JDBC了，因为ORM框架可以帮助我们更高效的进行开发。

学习Trident也一样，可能我们学习理解许多新的概念，但是学会了会极大的提高我们的开发效率。

Storm原语中，最重要的就是Spout、Bolt、Grouping等概念。

Trident对于Storm原语的抽象主要也就是针对这些基本概念的抽象。主要体现在：Trident Spout,Operation、State 。

Trident Spout是针对Storm原语中的Spout进行的抽象

Operation是针对Bolt、Grouping等概念的抽象

State是新提出的概念，实际上就是数据持久化的接口。

通常情况下，新的概念意味着要使用新的API。但是归根结底，还是底层还是通过storm原语来实现。在Trident中，我们使用TridentTopology表示一个拓扑，而在Storm原语中，我们使用StormTopology来表示一个拓扑。TridentTopology最终会被转换成StormTopology。在接下来的内容，我们将首先介绍TridentTopology的构建过程，以及TridentTopology如何转化为StormTopology。

二、TridentTopology与StormToplogy

1、API区别：

在Trident中，有着新的一套构建Topology的API，我们先通过从代码层面上对比来进行分析：

StormTopology：由传统的Spout和Bolt的API编写的Topology，最终是通过`TopologyBuilder`对象来创建的，返回的结果就是StormTopology对象。

StormTopologytopology= topologyBuilder.createTopology();

TridentTopology：由Trident的API编写的Topology，因为在Trident的API中，使用TridentTopology来表示一个Topology，是直接new出来的。

TridentTopology tridentTopology = new TridentTopology();

单独提出这两个概念，是为了以后的区分。以后我们提到StormTopology时，表示的就是以Spout、Bolt等这些API创建的Topology，而提到TridentTopology表示的就是以Trident的API创建的Topology。

2、创建Topology的区别

我们以单词计数案例WordCountApp(超链接)进行对比

StormTopology

TopologyBuilder builder = new TopologyBuilder();   builder.setSpout("word-reader" , new WordReader(),4);   builder.setBolt("word-normalizer" , new WordNormalizer(),3).shuffleGrouping("word-reader" );   builder.setBolt("word-counter" , new WordCounter(),1).fieldsGrouping("word-normalizer" , new Fields("word"));   StormTopology topology = builder .createTopology();

TridentTopology：

TridentTopology tridentTopology = new TridentTopology();              tridentTopology.newStream("word-reader-stream" , new WordReader()).parallelismHint(16)             .each( new Fields("line" ), new NormalizeFunction(), new Fields("word" ))             .groupBy( new Fields("word" ))             .persistentAggregate( new MemoryMapState.Factory(), new Sum(), new Fields("sum" )); StormTopology stormTopology = tridentTopology.build();

对比：

在StormTopology中，我们都是通过TopologyBuilder的setSpout、setBolt的方式来创建Topology，然后通过Grouping策略指定Bolt的数据来源和分组策略。

在TridentTopology中，我们使用TridentTopology来创建Topology，整个创建过程中，都是流式编程风格的。要注意的是，在Trident中，我们依然使用了WordReader这个Spout，但是并没有使用Bolt，而是使用了类似于each、persistentAggregate这样方法，来取代Bolt的功能。关于这些方法的作用再之后会详细介绍，目前只要知道Bolt的作用被一些方法取代了即可。

三、TridentTopology与StormToplogy的联系

二者的联系主要是：TridentTopology最终会被编译成StormTopology。请再次查看上述构建构建TridentTopology的最后一句代码。

StormTopology stormTopology = tridentTopology.build();

这句代码的的返回结果还是StormTopology对象，这实际上意味着，TridentTopology最终还是会被编译成StandardTopology。这很容易理解，就像ORM框架与JDBC一样，ORM框架只是一层封装，最终还是要通过JDBC操作数据库。而TridentTopology是高级封装，但是最终还是要通过编译StormTopology来运行。

注意：这一点是非常重要的。上面我们已经提到，在Trident中，依然要指定Spout，但是用了一系列其他的方法如each、persistentAggregate等（当然不止这些），代替了Bolt的功能。那么这里又提到，TridentTopology会被编译成StormTopology，实际上就意味着Storm最终会将这些方法转换为一个或多个Bolt。我们要了解Trident是如何工作的，就必须要了解，这些方法的最终是如何被转换为Bolt的。最简单的查看方式，就是查看编译后的StormTopology的getSpout，getBolt方法来看。

在后面我们会详细介绍，TridentTopology是如何转换为StormTopology的。目前，我们只需要知道TridentTopology最终是会转换为StormTopology即可。

事实上，在Trident框架会将调用的所有方法转换为一个个Node。Node类型如下：

newStream方法中参数Spout转换为SpoutNode，将调用的each方法，persistentAggregate等方法，转换为一个个ProcesserNode，而groupBy等操作，转换为一个PartitionNode，最终组成一个对象图，最后根据这个对象图，来将TridentTopology转换为StormTopology。

*  当然，光说不练假把式，我们通过分析源码进行简单说明。
 *  首先说明Spout转换为SpoutNode对象
 *  其次说明each、persistentAggregate转换为ProcesserNode
 *  最后说明如何根据SpoutNode和ProcesserNode将TridentTopology转换为StormTopology

1、Spout转换为SpoutNode

首先，我们看一下TridentTopology对象的newStream方法：

![Image.png][]

可以看到，可以接受五种类型的Spout，以下是这五个方法的实现：

![Image.png][Image.png 1]

我们可以看到，这五种方法，最终调用的实际上只有两个，并且在这两个方法中，最终都将Spout转换为了`SpoutNode`对象。

2、each、persistentAggregate转换为ProcessorNode

我们可以查看Stream对象的源码， 找到each、persistentAggregate两个方法内容

![Image.png][Image.png 2]

上图显示了这两个方法，这种都被转换为一个`ProcessorNode`，最终添加到Topology中。

3、最后我们看一下，TridentTopology.build()的实现

由于源码内容比较多，我们只分析感兴趣的地方，其中红色加深的地方，是目前最为关注的：

public StormTopology build() { ... TridentTopologyBuilder builder = new TridentTopologyBuilder();                 Map<Node, String> spoutIds = genSpoutIds( spoutNodes);         Map<Group, String> boltIds = genBoltIds( mergedGroups);         // SpoutNode维护了Spout类型，根据类型转换为对应的Spout         for(SpoutNode sn : spoutNodes ) {             Integer parallelism = parallelisms.get(grouper .nodeGroup(sn));             if(sn .type == SpoutNode.SpoutType.DRPC) {                 builder.setBatchPerTupleSpout(spoutIds .get(sn), sn.streamId ,                         (IRichSpout) sn. spout, parallelism , batchGroupMap.get(sn ));             } else {                 ITridentSpout s;                 if(sn .spout instanceof IBatchSpout) {                     s = new BatchSpoutExecutor((IBatchSpout)sn .spout );                 } else if(sn .spout instanceof ITridentSpout) {                     s = (ITridentSpout) sn. spout;                 } else {                     throw new RuntimeException("Regular rich spouts not supported yet... try wrapping in a RichSpoutBatchExecutor");                     // TODO: handle regular rich spout without batches (need lots of updates to support this throughout)                 }                 builder.setSpout(spoutIds .get(sn), sn.streamId, sn. txId, s, parallelism , batchGroupMap .get(sn));             }         }                 for(Group g : mergedGroups ) {             if(!isSpoutGroup( g)) {                 Integer p = parallelisms.get(g );                 Map<String, String> streamToGroup = getOutputStreamBatchGroups(g, batchGroupMap);                 //将调用each、processAggregate方法后的ProcessorNode，转换为Bolt                 BoltDeclarer d = builder.setBolt(boltIds .get(g), new SubtopologyBolt(graph, g .nodes , batchGroupMap ), p,                         committerBatches(g, batchGroupMap), streamToGroup);                 Collection<PartitionNode> inputs = uniquedSubscriptions(externalGroupInputs(g ));             //根据调用GroupBy等方法转换成的PartitionNode进行Grouping策略                 for(PartitionNode n : inputs ) {                     Node parent = TridentUtils.getParent( graph, n );                     String componentId;                     if(parent instanceof SpoutNode) {                         componentId = spoutIds .get(parent);                     } else {                         componentId = boltIds.get(grouper .nodeGroup(parent));                     }                     d.grouping( new GlobalStreamId(componentId , n.streamId ), n.thriftGrouping);                 }             }         }                 return builder .buildTopology(); }

一、认识storm trident  
    trident可以理解为storm批处理的高级抽象,提供了分组、分区、聚合、函数等操作,提供一致性和恰好一次处理的语义。  
    1)元祖被作为batch处理  
    2)每个batch的元祖都被指定唯一的一个事物id,如果因为处理失败导致batch重发,也和保证和重发前一样的事物id  
    3)数据更新操作严格有序,比如batch1必须在batch2之前被成功处理,且如果batch1失败了,后面的处理也会失败。  
    假如: batch1处理1--20  
          batch2处理21--40  
          batch1处理失败，那么batch2也会失败  
    虽然数据更新操作严格有序，但是数据处理阶段也可以并行的,只是最后的持久化操作必须有序。  
  1.1 trident state  
     trident的状态具有仅仅处理一次,持续聚合的语义,使用trident来实现恰好一次的语义不需要开发人员去处理事务相关的工作,因为trident state已经帮我们封装好了,只需要编写类似于如下的代码：