Storm Trident API 实践

标签：storm trident   trident api   trident api 实践   storm trident api 详解   

一、概要     

1.1 Storm(简介)

1.2 Trident(简介)

二、Trident API 实践

2.1 each() 方法

2.1.1 Filter类：过滤tuple

public static class PerActorTweetsFilter extends BaseFilter {
  String actor;

  public PerActorTweetsFilter(String actor) {
    this.actor = actor;
  }
  @Override
  public boolean isKeep(TridentTuple tuple) {
    return tuple.getString(0).equals(actor);
  }
}

topology.newStream("spout", spout)
  .each(new Fields("actor", "text"), new PerActorTweetsFilter("dave"))
  .each(new Fields("actor", "text"), new Utils.PrintFilter());

1. 第一个构造参数：作为Field Selector，一个tuple可能有很多字段，通过设置Field，我们可以隐藏其它字段，仅仅接收指定的字段（其它字段实际还在）。
2. 第二个是一个Filter：用来过滤掉除actor名叫"dave"外的其它消息。

2.1.2 Function类：加工处理tuple内容

public static class UppercaseFunction extends BaseFunction {
  @Override
  public void execute(TridentTuple tuple, TridentCollector collector) {
    collector.emit(new Values(tuple.getString(0).toUpperCase()));
  }
}

topology.newStream("spout", spout)
  .each(new Fields("actor", "text"), new PerActorTweetsFilter("dave"))
  .each(new Fields("text", "actor"), new UppercaseFunction(), new Fields("uppercased_text"))
  .each(new Fields("actor", "text", "uppercased_text"), new Utils.PrintFilter());

2.1.3 Field Selector与project

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

2.2 parallelismHint()方法和partitionBy()

2.2.1 parallelismHint()

public static class PerActorTweetsFilter extends BaseFilter {

  private int partitionIndex;
  private String actor;

  public PerActorTweetsFilter(String actor) {
    this.actor = actor;
  }
  @Override
  public void prepare(Map conf, TridentOperationContext context) {
    this.partitionIndex = context.getPartitionIndex();
  }
  @Override
  public boolean isKeep(TridentTuple tuple) {
    boolean filter = tuple.getString(0).equals(actor);
    if(filter) {
      System.err.println("I am partition [" + partitionIndex + "] and I have kept a tweet by: " + actor);
    }
    return filter;
  }
}

topology.newStream("spout", spout)
  .each(new Fields("actor", "text"), new PerActorTweetsFilter("dave"))
  .parallelismHint(5)
  .each(new Fields("actor", "text"), new Utils.PrintFilter());

I am partition [4] and I have kept a tweet by: dave
I am partition [3] and I have kept a tweet by: dave
I am partition [0] and I have kept a tweet by: dave
I am partition [2] and I have kept a tweet by: dave
I am partition [1] and I have kept a tweet by: dave

topology.newStream("spout", spout)
  .parallelismHint(2)
  .shuffle()
  .each(new Fields("actor", "text"), new PerActorTweetsFilter("dave"))
  .parallelismHint(5)
  .each(new Fields("actor", "text"), new Utils.PrintFilter());

2.2.2 partitionBy()和重定向操作(repartitioning operation)  

I am partition [2] and I have kept a tweet by: dave
I am partition [2] and I have kept a tweet by: dave
I am partition [2] and I have kept a tweet by: dave
I am partition [2] and I have kept a tweet by: dave

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 聚合(Aggregation)

2.3.1 Aggregator:

public static class LocationAggregator extends BaseAggregator<Map<String, Integer>> {

  @Override
  public Map<String, Integer> init(Object batchId, TridentCollector collector) {
    return new HashMap<String, Integer>();
  }

  @Override
  public void aggregate(Map<String, Integer> val, TridentTuple tuple, TridentCollector collector) {
    String location = tuple.getString(0);
    val.put(location, MapUtils.getInteger(val, location, 0) + 1);
  }

  @Override
  public void complete(Map<String, Integer> val, TridentCollector collector) {
    collector.emit(new Values(val));
  }
}

topology.newStream("spout", spout)
  .aggregate(new Fields("location"), new LocationAggregator(), new Fields("location_counts"))
  .each(new Fields("location_counts"), new Utils.PrintFilter());

1. init():当刚开始接收到一个batch时执行
2. aggregate():在接收到batch中的每一个tuple时执行
3. complete():在一个batch的结束时执行     

[{USA=3, Spain=1, UK=1}]
[{USA=3, Spain=2}]
[{France=1, USA=4}]
[{USA=4, Spain=1}]
[{USA=5}]

2.3.2 partitionAggregate():

     如果我们将上面的Topology稍微改造一下，猜一下结果会是如何？

topology.newStream("spout", spout)
  .partitionBy(new Fields("location"))
  .partitionAggregate(new Fields("location"), new LocationAggregator(), new Fields("location_counts"))
  .parallelismHint(3)
  .each(new Fields("location_counts"), new Utils.PrintFilter());

[{France=10, Spain=5}]
[{USA=63}]
[{UK=22}]

2.4 groupBy

topology.newStream("spout", spout)
  .groupBy(new Fields("location"))
  .aggregate(new Fields("location"), new Count(), new Fields("count"))
  .each(new Fields("location", "count"), new Utils.PrintFilter());

...
[France, 25]
[UK, 2]
[USA, 25]
[Spain, 44]
[France, 26]
[UK, 3]
...

     不过需要注意的是，groupBy()本身并不是一个重定向操作，但如果它后面跟的是aggregator的话就是，跟的是partitionAggregate的话就不是。

三、总结 

作者：suifeng3051
原文地址：http://blog.csdn.net/suifeng3051/article/details/41118721

Storm Trident API 实践

标签：storm trident   trident api   trident api 实践   storm trident api 详解   

原文地址：http://blog.csdn.net/suifeng3051/article/details/41118721