trident苍蓝?trident什么意思?Storm Trident 详细介绍(2)

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

如果我们指定执行Filter任务的线程数量为5，那么最终的执行结果会如何呢？看一下我们的测试结果：

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

我们可以很清楚的发现，一共有5个线程在执行Filter。

如果我们想要2个Spout和5个Filter怎么办呢？如下面代码所示，实现很简单。

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)

我们注意到上面的例子中用到了shuffle()，shuffle()是一个重定向操作。那什么是重定向操作呢？重定向定义了我们的tuple如何被route到下一处理层，当然不同的层之间可能会有不同的并行度，shuffle()的作用是把tuple随机的route下一层的线程中，而partitionBy()则根据我们的指定字段按照一致性哈希算法route到下一层的线程中，也就是说，如果我们用partitionBy()的话，同一个字段名的tuple会被route到同一个线程中。

比如，如果我们把上面代码中的shuffle()改成partitionBy(new Fields("actor"))，猜一下结果会怎样？

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

测试结果正如我们上面描述的那样，相同字段的tuple被route到了同一个partition中。

重定向操作有如下几种：

2.3 聚合(Aggregation)

我们前面讲过，Trident的一个很重要的特点就是它是以batch的形式处理tuple的。我们可以很容易想到的针对一个batch的最基本操作应该就是聚合。trident提供了聚合API来处理batches，来看一个例子：

2.3.1Aggregator:

public static class LocationAggregator extends BaseAggregator

以上就是关于trident的全部内容，相信你一定会非常满意。

本文来自电脑杂谈，转载请注明本文网址：
https://www.pc-fly.com/a/shenmilingyu/article-16667-2.html