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

你是否正在寻找关于trident的内容？让我把最内涵的东西奉献给你：

1.1 Storm(简介)

Storm是一个实时的可靠地分布式流计算框架。

具体就不多说了，举个例子，它的一个典型的大数据实时计算应用场景：从Kafka消息队列读取消息（可以是logs,clicks,sensor data）、通过Storm对消息进行计算聚合等预处理、把处理结果持久化到NoSQL或者HDFS做进一步深入分析。

1.2 trident(简介)

trident是对Storm的更高一层的抽象,除了提供一套简单易用的流数据处理API之外，它以batch(一组tuples)为单位进行处理，这样一来，可以使得一些处理更简单和高效，。

我们知道把Bolt的运行状态仅仅保存在内存中是不可靠的，如果一个node挂掉，那么这个node上的任务就会被重新分配，但是之前的状态是无法恢复的。因此，比较聪明的方式就是把storm的计算状态信息持久化到database中，基于这一点，trident就变得尤为重要。因为在处理大数据时，我们在与database打交道时通常会采用批处理的方式来避免给它带来压力，而trident恰恰是以batch groups的形式处理数据，并提供了一些聚合功能的API。

二、trident API 实践

Trident其实就是一套API，但现阶段网上关于Trident API中各个函数的用法含义资料不多，下面我就根据一些英文资料和自己的理解，详细介绍一下trident API各个函数的用法和含义。

2.1 each() 方法

作用：操作batch中的每一个tuple内容，一般与Filter或者Function函数配合使用。

下面通过一个例子来介绍each()方法，假设我们有一个FakeTweetsBatchSpout，它会模拟一个Stream，随机产生一个个消息。我们可以通过设置这个Spout类的构造参数来改变这个Spout的batch Size的大小。

2.1.1 Filter类：过滤tuple

一个通过actor字段过滤消息的Filter：

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：

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

从上面例子看到，each()方法有一些构造参数

2.1.2 Function类：加工处理tuple内容

一个能把tuple中text内容变成大写的Function：

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

Topology：

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

首先，UppercaseFunction函数的输入是Fields("text","actor")，其作用是把其中的"text"字段内容都变成大写。

2.1.3 Field Selector与project

我们需要注意的是，上面每个each()方法的第一个Field字段仅仅是隐藏掉没有指定的字段内容，实际上被隐藏的字段依然还在tuple中，如果想要彻底丢掉它们，我们就需要用到project()方法。

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

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

2.2 parallelismHint()方法和partitionBy()

2.2.1 parallelismHint()

指定Topology的并行度，即用多少线程执行这个任务。我们可以稍微改一下我们的Filter，通过打印当前任务的partitionIndex来区分当前是哪个线程。

Filter：

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：

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