trident原理及编程指南

@(STORM)[storm, 大数据]

trident原理及编程指南
一理论介绍
- 一trident是什么
- 二trident处理单位
- 三事务类型
  - 1spout类型
  - 2state类型
  - 3实现恰好一次的spout与state组合类型
二编程指南
- 1定义输入流
- 2统计单词数量
- 3输出统计结果
- 4split的字义
三使用kafka作为数据源
- 1定义kafka相关配置
- 2从kafka中读取消息并处理
- 3提交拓扑
四State示例
- 1主类
- 2Aggregator的用法
  - 1Aggregator接口
  - 2init方法
  - 3aggregate方法
  - 4complete方法
- 3state的用法
  - 1拓扑定义
  - 2工厂类NameSumStateFactory
  - 3更新类NameSumUpdater
  - 4状态类NameSumState
- 4state应用思路总结

一、理论介绍

（一）trident是什么？

Trident is a high-level abstraction for doing realtime computing on top of Storm. It allows you to seamlessly intermix high throughput (millions of messages per second), stateful stream processing with low latency distributed querying. If you’re familiar with high level batch processing tools like Pig or Cascading, the concepts of Trident will be very familiar – Trident has joins, aggregations, grouping, functions, and filters. In addition to these, Trident adds primitives for doing stateful, incremental processing on top of any database or persistence store. Trident has consistent, exactly-once semantics, so it is easy to reason about Trident topologies.
简单的说，trident是storm的更高层次抽象，相对storm，它主要提供了3个方面的好处：
（1）提供了更高层次的抽象，将常用的count,sum等封装成了方法，可以直接调用，不需要自己实现。
（2）以批次代替单个元组，每次处理一个批次的数据。
（3）提供了事务支持，可以保证数据均处理且只处理了一次。

（二）trident处理单位

trident每次处理消息均为batch为单位，即一次处理多个元组。

（三）事务类型

关于事务类型，有2个比较容易混淆的概念：spout的事务类型以及事务状态。
它们都有3种类型，分别为：事务型、非事务型和透明事务型。

1、spout类型

spout的类型指定了由于下游出现问题（fail被调用，或者超时无回复）导致元组需要重放时，应该怎么发送元组。
事务型spout:重放时能保证同一个批次发送同一批元组。可以保证每一个元组都被发送且只发送一个，且同一个批次所发送的元组是一样的。
非事务型spout：没有任何保障，发完就算。
透明事务型spout：同一个批次发送的元组有可能不同的，它可以保证每一个元组都被发送且只发送一次，但不能保证重放时同一个批次的数据是一样的。这对于部分失效的情况尤其有用，假如以kafka作为spout，当一个topic的某个分区失效时，可以用其它分区的数据先形成一个批次发送出去，如果是事务型spout，则必须等待那个分区恢复后才能继续发送。
这三种类型可以分别通过实现ITransactionalSpout、ITridentSpout、IOpaquePartitionedTridentSpout接口来定义。

2、state类型

state的类型指定了如果将storm的中间输出或者最终输出持久化到某个地方（如内存），当某个批次的数据重放时应该如果更新状态。state对于下游出现错误的情况尤其有用。
事务型状态：同一批次tuple提供的结果是相同的。
非事务型状态：没有回滚能力，更新操作是永久的。
透明事务型状态：更新操作基于先前的值，这样由于这批数据发生变化，对应的结果也会发生变化。透明事务型状态除了保存当前数据外，还要保存上一批数据，当数据重放时，可以基于上一批数据作更新。
注意，此处的状态应该是原子性的，比如将状态写入hbase，则应该全部写入，或者全部没写入，不能说写入一半，另一半没写入，这连事务型也无法保证恰好一次了。比如说写入本地磁盘，就有可能导致这种情况，如果写到一半出错，则无法保证恰好一次了，因为磁盘没有类似于数据库的commit、rollback操作。

3、实现恰好一次的spout与state组合类型

由上表可以看出：
（1）当spout与state均为transcational或者均为opaque时，可以实现恰好一次。
（2）当spout为tansactional，state为opaque时，也可以实现恰好一次。
（3）但当spout为opaque，state为transactional时，不可以实现恰好一次。因此opaque spout重发时，它的内容可能与上一次不同，而state如果在上个批次已经更新过但这个批次最终fail了，则spout重发时，会在已经fail掉的批次上更新，而上一个批次是不应该计算在内的。如果state是transactional的，则它同时保存了上一次状态及当前状态，所以可以基于上一次的状态作更新，就不会有这个问题。

二、编程指南

代码如下

package org.ljh.tridentdemo;import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.FilterNull;
import storm.trident.operation.builtin.MapGet;
import storm.trident.operation.builtin.Sum;
import storm.trident.testing.FixedBatchSpout;
import storm.trident.testing.MemoryMapState;
import storm.trident.tuple.TridentTuple;public class TridentWordCount {public static class Split extends BaseFunction {@Overridepublic void execute(TridentTuple tuple, TridentCollector collector) {String sentence = tuple.getString(0);for (String word : sentence.split(" ")) {collector.emit(new Values(word));}}}public static StormTopology buildTopology(LocalDRPC drpc) {FixedBatchSpout spout =new FixedBatchSpout(new Fields("sentence"), 3, new Values("the cow jumped over the moon"), new Values("the man went to the store and bought some candy"), new Values("four score and seven years ago"),new Values("how many apples can you eat"), new Values("to be or not to be the person"));spout.setCycle(true);//创建拓扑对象TridentTopology topology = new TridentTopology();//这个流程用于统计单词数据，结果将被保存在wordCounts中TridentState wordCounts =topology.newStream("spout1", spout).parallelismHint(16).each(new Fields("sentence"), new Split(), new Fields("word")).groupBy(new Fields("word")).persistentAggregate(new MemoryMapState.Factory(), new Count(),new Fields("count")).parallelismHint(16);//这个流程用于查询上面的统计结果topology.newDRPCStream("words", drpc).each(new Fields("args"), new Split(), new Fields("word")).groupBy(new Fields("word")).stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count")).each(new Fields("count"), new FilterNull()).aggregate(new Fields("count"), new Sum(), new Fields("sum"));return topology.build();}public static void main(String[] args) throws Exception {Config conf = new Config();conf.setMaxSpoutPending(20);if (args.length == 0) {LocalDRPC drpc = new LocalDRPC();LocalCluster cluster = new LocalCluster();cluster.submitTopology("wordCounter", conf, buildTopology(drpc));for (int i = 0; i < 100; i++) {System.out.println("DRPC RESULT: " + drpc.execute("words", "cat the dog jumped"));Thread.sleep(1000);}} else {conf.setNumWorkers(3);StormSubmitter.submitTopologyWithProgressBar(args[0], conf, buildTopology(null));}}
}

实例实现了最基本的wordcount功能，然后将结果输出。关键步骤如下：

1、定义输入流

    FixedBatchSpout spout =new FixedBatchSpout(new Fields("sentence"), 3, new Values("the cow jumped over the moon"), new Values("the man went to the store and bought some candy"), new Values("four score and seven years ago"),new Values("how many apples can you eat"), new Values("to be or not to be the person"));spout.setCycle(true);

（1）使用FixedBatchSpout创建一个输入spout，spout的输出字段为sentence，每3个元组作为一个batch。
（2）数据不断的重复发送。

2、统计单词数量

    TridentState wordCounts =topology.newStream("spout1", spout).parallelismHint(16).each(new Fields("sentence"), new Split(), new Fields("word")).groupBy(new Fields("word")).persistentAggregate(new MemoryMapState.Factory(), new Count(),new Fields("count")).parallelismHint(16);

这个流程用于统计单词数据，结果将被保存在wordCounts中。6行代码的含义分别为：
（1）首先从spout中读取消息，spout1定义了zookeeper中用于保存这个拓扑的节点名称。
（2）并行度设置为16，即16个线程同时从spout中读取消息。
（3）each中的三个参数分别为：输入字段名称，处理函数，输出字段名称。即从字段名称叫sentence的数据流中读取数据，然后经过new Split()处理后，以word作为字段名发送出去。其中new Split()后面介绍，它的功能就是将输入的内容以空格为界作了切分。
（4）将字段名称为word的数据流作分组，即相同值的放在一组。
（5）将已经分好组的数据作统计，结果放到MemoryMapState，然后以count作为字段名称将结果发送出去。这步骤会同时存储数据及状态，并将返回TridentState对象。
（6）并行度设置。

3、输出统计结果

    topology.newDRPCStream("words", drpc).each(new Fields("args"), new Split(), new Fields("word")).groupBy(new Fields("word")).stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count")).each(new Fields("count"), new FilterNull()).aggregate(new Fields("count"), new Sum(), new Fields("sum"));

这个流程从上述的wordCounts对象中读取结果，并返回。6行代码的含义分别为：

（1）等待一个drpc调用，从drpc服务器中接受words的调用来提供消息。调用代码如下：

drpc.execute("words", "cat the dog jumped")

（2）输入为上述调用中提供的参数，经过Split()后，以word作为字段名称发送出去。
（3）以word的值作分组。
（4）从wordCounts对象中查询结果。4个参数分别代表：数据来源，输入数据，内置方法（用于从map中根据key来查找value），输出名称。
（5）过滤掉空的查询结果，如本例中，cat和dog都没有结果。
（6）将结果作统计，并以sum作为字段名称发送出去，这也是DRPC调用所返回的结果。如果没有这一行，最后的输出结果

DRPC RESULT: [["cat the dog jumped","the",2310],["cat the dog jumped","jumped",462]]

加上这一行后，结果为：

DRPC RESULT: [[180]]

4、split的字义

public static class Split extends BaseFunction {@Overridepublic void execute(TridentTuple tuple, TridentCollector collector) {String sentence = tuple.getString(0);for (String word : sentence.split(" ")) {collector.emit(new Values(word));}}
}

注意它最后会发送数据。

5、创建并启动拓扑

public static void main(String[] args) throws Exception {Config conf = new Config();conf.setMaxSpoutPending(20);if (args.length == 0) {LocalDRPC drpc = new LocalDRPC();LocalCluster cluster = new LocalCluster();cluster.submitTopology("wordCounter", conf, buildTopology(drpc));for (int i = 0; i < 100; i++) {System.out.println("DRPC RESULT: " + drpc.execute("words", "cat the dog jumped"));Thread.sleep(1000);}} else {conf.setNumWorkers(3);StormSubmitter.submitTopologyWithProgressBar(args[0], conf, buildTopology(null));}
}

（1）当无参数运行时，启动一个本地的集群，及自已创建一个drpc对象来输入。
（2）当有参数运行时，设置worker数量为3，然后提交拓扑到集群，并等待远程的drpc调用。

三、使用kafka作为数据源

package com.netease.sytopology;import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Arrays;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import storm.kafka.BrokerHosts;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import storm.kafka.trident.OpaqueTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentTopology;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.operation.builtin.Count;
import storm.trident.testing.MemoryMapState;
import storm.trident.tuple.TridentTuple;
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;/** 本类完成以下内容*/
public class SyTopology {public static final Logger LOG = LoggerFactory.getLogger(SyTopology.class);private final BrokerHosts brokerHosts;public SyTopology(String kafkaZookeeper) {brokerHosts = new ZkHosts(kafkaZookeeper);}public StormTopology buildTopology() {TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(brokerHosts, "ma30", "storm");kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());// TransactionalTridentKafkaSpout kafkaSpout = new// TransactionalTridentKafkaSpout(kafkaConfig);OpaqueTridentKafkaSpout kafkaSpout = new OpaqueTridentKafkaSpout(kafkaConfig);TridentTopology topology = new TridentTopology();// TridentState wordCounts =topology.newStream("kafka4", kafkaSpout).each(new Fields("str"), new Split(),new Fields("word")).groupBy(new Fields("word")).persistentAggregate(new MemoryMapState.Factory(), new Count(),new Fields("count")).parallelismHint(16);// .persistentAggregate(new HazelCastStateFactory(), new Count(),// new Fields("aggregates_words")).parallelismHint(2);return topology.build();}public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {String kafkaZk = args[0];SyTopology topology = new SyTopology(kafkaZk);Config config = new Config();config.put(Config.TOPOLOGY_TRIDENT_BATCH_EMIT_INTERVAL_MILLIS, 2000);String name = args[1];String dockerIp = args[2];config.setNumWorkers(9);config.setMaxTaskParallelism(5);config.put(Config.NIMBUS_HOST, dockerIp);config.put(Config.NIMBUS_THRIFT_PORT, 6627);config.put(Config.STORM_ZOOKEEPER_PORT, 2181);config.put(Config.STORM_ZOOKEEPER_SERVERS, Arrays.asList(dockerIp));StormSubmitter.submitTopology(name, config, topology.buildTopology());}static class Split extends BaseFunction {public void execute(TridentTuple tuple, TridentCollector collector) {String sentence = tuple.getString(0);for (String word : sentence.split(",")) {try {FileWriter fw = new FileWriter(new File("/home/data/test/ma30/ma30.txt"),true);fw.write(word);fw.flush();fw.close();} catch (IOException e) {// TODO Auto-generated catch blocke.printStackTrace();}collector.emit(new Values(word));}}}
}

本例将从kafka中读取消息，然后对消息根据“，”作拆分，并写入一个本地文件。

1、定义kafka相关配置

    TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(brokerHosts, "ma30", "storm");kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());OpaqueTridentKafkaSpout kafkaSpout = new OpaqueTridentKafkaSpout(kafkaConfig);

其中ma30是订阅的topic名称。

2、从kafka中读取消息并处理

    topology.newStream("kafka4", kafkaSpout).each(new Fields("str"), new Split(),new Fields("word")).groupBy(new Fields("word")).persistentAggregate(new MemoryMapState.Factory(), new Count(),new Fields("count")).parallelismHint(16);

（1）指定了数据来源，并指定zookeeper中用于保存数据的位置，即保存在/transactional/kafka4。
（2）指定处理方法及发射的字段。
（3）根据word作分组。
（4）计数后将状态写入MemoryMapState

3、提交拓扑：

storm jar target/sytopology2-0.0.1-SNAPSHOT.jar com.netease.sytopology.SyTopology 192.168.172.98:2181/kafka test3 192.168.172.98

此时可以在/home/data/test/ma30/ma30.txt看到split的结果

四、State示例

trident通过spout的事务性与state的事务处理，保证了恰好一次的语义。这里介绍了如何使用state。

完整代码请见 https://github.com/lujinhong/tridentdemo

1、主类

主类定义了拓扑的整体逻辑，这个拓扑通过一个固定的spout循环产生数据，然后统计消息中每个名字出现的次数。

拓扑中先将消息中的内容提取出来成name, age, title, tel4个field，然后通过project只保留name字段供统计，接着按照name分区后，为每个分区进行聚合，最后将聚合结果通过state写入map中。

    storm.trident.Stream Origin_Stream = topology.newStream("tridentStateDemoId", spout).parallelismHint(3).shuffle().parallelismHint(3).each(new Fields("msg"), new Splitfield(),new Fields("name", "age", "title", "tel")).parallelismHint(3).project(new Fields("name")) //其实没什么必要，上面就不需要发射BCD字段，但可以示范一下project的用法.parallelismHint(3).partitionBy(new Fields("name"));   //根据name的值作分区Origin_Stream.partitionAggregate(new Fields("name"), new NameCountAggregator(),new Fields("nameSumKey", "nameSumValue")).partitionPersist(new NameSumStateFactory(), new Fields("nameSumKey", "nameSumValue"),new NameSumUpdater());

2、Aggregator的用法

这里涉及了一些trident常用的API，但project等相对容易理解，这里只介绍partitionAggregate的用法。

再看看上面代码中对partitionAggregate的使用：

Origin_Stream.partitionAggregate(new Fields("name"), new NameCountAggregator(),new Fields("nameSumKey", "nameSumValue"))

第一，三个参数分别表示输入流的名称与输出流的名称。中间的NameCountAggregator是一个Aggregator的对象，它定义了如何对输入流进行聚合。我们看一下它的代码：

public class NameCountAggregator implements Aggregator<Map<String, Integer>> {private static final long serialVersionUID = -5141558506999420908L;@Overridepublic Map<String, Integer> init(Object batchId,TridentCollector collector) {return new HashMap<String, Integer>();}//判断某个名字是否已经存在于map中，若无，则put，若有，则递增@Overridepublic void aggregate(Map<String, Integer> map,TridentTuple tuple, TridentCollector collector) {String key=tuple.getString(0);if(map.containsKey(key)){Integer tmp=map.get(key);map.put(key, ++tmp);}else{map.put(key, 1);}}//将聚合后的结果emit出去@Overridepublic void complete(Map<String, Integer> map,TridentCollector collector) {if (map.size() > 0) {for(Entry<String, Integer> entry : map.entrySet()){System.out.println("Thread.id="+Thread.currentThread().getId()+"|"+entry.getKey()+"|"+entry.getValue());collector.emit(new Values(entry.getKey(),entry.getValue()));}map.clear();} }@Overridepublic void prepare(Map conf, TridentOperationContext context) {}@Overridepublic void cleanup() {}}

（1）Aggregator接口

它实现了Aggregator接口，这个接口有3个方法：

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

init方法：在处理batch之前被调用。init的返回值是一个表示聚合状态的对象，该对象会被传递到aggregate和complete方法。
aggregate方法：为每个在batch分区的输入元组所调用，更新状态
complete方法：当batch分区的所有元组已经被aggregate方法处理完后被调用。

除了实现Aggregator接口，还可以实现ReducerAggregator或者CombinerAggregator，它们使用更方便。详见《从零开始学storm》或者官方文档
https://storm.apache.org/documentation/Trident-API-Overview.html

下面我们看一下这3个方法的实现。

（2）init方法

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

仅初始化了一个HashMap对象，这个对象会作为参数传给aggregate和complete方法。对一个batch只执行一次。

（3）aggregate方法

aggregate方法对于batch内的每一个tuple均执行一次。这里将这个batch内的名字出现的次数放到init方法所初始化的map中。

@Override
public void aggregate(Map<String, Integer> map,TridentTuple tuple, TridentCollector collector) {String key=tuple.getString(0);if(map.containsKey(key)){Integer tmp=map.get(key);map.put(key, ++tmp);}else{map.put(key, 1);}
}

（4）complete方法

这里在complete将aggregate处理完的结果发送出去，实际上可以在任何地方emit，比如在aggregate里面。
这个方法对于一个batch也只执行一次。

@Override
public void complete(Map<String, Integer> map,TridentCollector collector) {if (map.size() > 0) {for(Entry<String, Integer> entry : map.entrySet()){System.out.println("Thread.id="+Thread.currentThread().getId()+"|"+entry.getKey()+"|"+entry.getValue());collector.emit(new Values(entry.getKey(),entry.getValue()));}map.clear();}
}

3、state的用法

（1）拓扑定义

先看一下主类中如何将结果写入state：

partitionPersist(new NameSumStateFactory(), new Fields("nameSumKey", "nameSumValue"),new NameSumUpdater());

它的定义为：

TridentState storm.trident.Stream.partitionPersist(StateFactory stateFactory, Fields inputFields, StateUpdater updater)

其中的第二个参数比较容易理解，就是输入流的名称，这里是名字与它出现的个数。下面先看一下Facotry。

（2）工厂类：NameSumStateFactory

很简单，它实现了StateFactory，只有一个方法makeState，返回一个State类型的对象。

public class NameSumStateFactory implements StateFactory {private static final long serialVersionUID = 8753337648320982637L;@Overridepublic State makeState(Map arg0, IMetricsContext arg1, int arg2, int arg3) {return new NameSumState();  }
}

（3）更新类：NameSumUpdater

这个类继承自BaseStateUpdater，它的updateState对batch的内容进行处理，这里是将batch的内容放到一个map中，然后调用setBulk方法

public class NameSumUpdater extends BaseStateUpdater<NameSumState> {private static final long serialVersionUID = -6108745529419385248L;public void updateState(NameSumState state, List<TridentTuple> tuples, TridentCollector collector) {Map<String,Integer> map=new HashMap<String,Integer>();for(TridentTuple t: tuples) {map.put(t.getString(0), t.getInteger(1));}state.setBulk(map);}
}

（4）状态类：NameSumState

这是state最核心的类，它实现了大部分的逻辑。NameSumState实现了State接口：

public interface State {void beginCommit(Long txid); void commit(Long txid);
}

分别在提交之前与提交成功的时候调用，在这里只打印了一些信息。

另外NameSumState还定义了如何处理NameSumUpdater传递的消息：

public void setBulk(Map<String, Integer> map) {// 将新到的tuple累加至map中for (Entry<String, Integer> entry : map.entrySet()) {String key = entry.getKey();if (this.map.containsKey(key)) {this.map.put(key, this.map.get(key) + map.get(key));} else {this.map.put(key, entry.getValue());}}System.out.println("-------");// 将map中的当前状态打印出来。for (Entry<String, Integer> entry : this.map.entrySet()) {String Key = entry.getKey();Integer Value = entry.getValue();System.out.println(Key + "|" + Value);}
}

即将NameSumUpdater传送过来的内容写入一个HashMap中，并打印出来。
此处将state记录在一个HashMap中，如果需要记录在其它地方，如mysql，则使用jdbc写入mysql代替下面的map操作即可。

事实上，这个操作不一定要在state中执行，可以在任何类中，但建议还是在state类中实现。

4、state应用思路总结

（1）使用state，你不再需要比较事务id，在数据库中同时写入多个值等内容，而是专注于你的逻辑实现
（2）除了实现State接口，更常用的是实现MapState接口，下次补充。
（3）在拓扑中指定了StateFactory，这个工厂类找到相应的State类。而Updater则每个批次均会调用它的方法。State中则定义了如何保存数据，这里将数据保存在内存中的一个HashMap，还可以保存在mysql, hbase等等。
（4）trident会自动比较txid的值，如果和当前一样，则不更改状态，如果是当前txid的下一个值，则更新状态。这种逻辑不需要用户处理。
（5）如果需要实现透明事务状态，则需要保存当前值与上一个值，在update的时候2个要同时处理。即逻辑由自己实现。在本例子中，大致思路是在NameSumState中创建2个HashMap，分别对应当前与上一个状态的值，而NameSumUpdater每次更新这2个Map。