storm笔记：Trident状态

在storm笔记：Trident应用中说了下Trident的使用，这里说下Trident几种状态的变化及其对应API的使用。

本文内容来自Trident State，部分内容根据实际情况做出修改。

Trident中有对状态数据进行读取和写入操作的一流抽象工具。状态既可以保存在拓扑内部，比如保存在内容中并由HDFS存储，也可以通过外部存储（比如Memcached或Cassandra）存储在数据库中。而对于Trident的API而言，这两种机制没有任何区别。

Trident以容错的方式管理状态，以便在重试或失败时的状态更新是幂等的。在大数据处理中，数据处理的幂等性是非常重要的一个指标，这样能够保证每个消息即使处理了多次，结果也像是只处理了一次一样。

在进行状态更新时可能需要各种级别的容错能力，在这之前，我们来看一个例子说明实现“恰好一次”语义所需的技巧。比如，正在对流中的数据进行计数聚合操作，每次处理新的元组时，都会将运行的计数结果存储在数据库中。

如果发生故障时，元组将重新执行计数操作。这就会在执行状态更新时出现问题，因为这个时候不知道是不是已经更新过该元组状态。也许还没有处理该元组数据，这个时候就需要增加计数。也许已经处理该元组，并成功增加计数，但是在下一步的时候出现问题，这种情况下，就不应该增加计数。也有可能是处理元组正常，更新计数是异常，这个时候就需要更新计数。

所以说，如果只是在数据库中存储计数信息，就不知道元组是否已经处理过。因此，就需要更多的信息作为辅助。Trident提供了下面三个性质，来实现“恰好一次”的处理：

元组都是以小批次处理

每批元组都会给出一个唯一ID，称为事务ID（transaction id，txid）。如果批次重复处理，txid也会相同。

状态的更新操作是按照元组批次的顺序执行的。也就是说，在批次2状态更新成功之前，不会进行批次3的状态更新。

根据这些特性，就可以通过检查到该元组的批次是否已被处理，并根据检测结果采取适当的操作更新状态了。采取的具体操作取决于Spout的类型。Spout有三种类型：“非事务型（non-transactional）”，“事务型（transactional）”和“不透明事务型（opaque transactional）”。对应的容错能力也是三种：“非事务”，“事务”和“不透明事务”。下面来看看Spout的各个类型及对应的容错能力。

事务型Spout

Trident是按照批次发送元组进行处理的，每个批次的元组被赋予唯一的事务ID。Spout的特性根据他们所提供容错性保证机制来决定的，而且这种机制也会对每个批次发生作用。事务型Spout有如下特性：

每个批次的txid不变，对于一个特定的txid，重复执行时，它所包含的元组数据与第一次完全相同。

元组只会在一个批次出现，不会重复（某个元组只会出现在一个批次中，不会出现在多个批次中）。

每个元组都会出现一次（不会遗漏任何的元组数据）

这是最简单最容易理解的一种Spout类型，数据流被分割成固定的批次。storm中有与Kafka集成的事务型Spout的扩展，代码在这里。

既然事务型Spout这么简单易懂，为什么不在Trident中完全使用事务型Spout呢？其实就在于它的容错能力。比如，TransactionalTridentKafkaSpout的工作方式是，同一个txid的批次中将包含kafka所有分区的元组。一旦某个批次发出后，出现异常，需要重新发出，就需要完全相同的元组集合才能满足事务型Spout要求的语义。但是这个时候，kafka某个节点异常（节点关闭或分区不可用），就无法获取完全相同的的一批元组，那整个拓扑就会应为第3条语义（批次按顺序执行）停止。

这就是要有“不透明事务型”Spout的原因了，它能够容忍数据源节点丢失，而且又能保证数据恰好被操作一次。

注：对kafka比较熟悉的应该会想到，如果某一个topic支持复制，那即使一个节点不可用，还会有其他复制节点顶上，那TransactionalTridentKafkaSpout也能够避免上面的问题。

下面继续看看如何设计一个支持恰好一次特性的“事务型”Spout语义（简单的说就是同一个txid对应的批次元组数据完全一致）的状态实现，这种状态称为“事务型状态”。

比如，现在有一个单词计数的拓扑，需要将单词计数存储在key/value数据库中。key是单词，value中包含单词数量。另外，为了确定同一批次元组是否已经被执行，需要将txid也一同存储在value中。这样，当需要更新单词数量的时候，先比较txid是否相同，如果相同，就跳过更新。如果不同，就更新计数。

考虑这个为什么它工作的例子。 假设您正在处理由以下批次元组组成的txid 3：

比如，要处理一个txid是3的一批元组：

["man"]
["man"]
["dog"]

目前数据库中存储的数据为：

man => [count=3, txid=1]
dog => [count=4, txid=3]
apple => [count=10, txid=2]

在这个时候，发现“man”对应的txid是1，当前的txid是3，就可以更新了。然后“dog”对应的txid是3，说明同一批次的元组数据已经发送过了，就不需要更新。从这点可以看出，txid是3的批次元组是重复发送的，在更新“dog”数量后，在更新“man”数量前，出现了错误。最后的结果就是：

man => [count=5, txid=3]
dog => [count=4, txid=3]
apple => [count=10, txid=2]

不透明事务型Spout

前面已经提过，不透明事务型Spout不能保证相同txid对应的批次中的元组数据完全一致。其特点如下：

每个元组都会在有且仅有一个批次中处理成功。

[OpaqueTridentKafkaSpout](http://github.com/apache/storm/tree/v1.1.0/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/OpaqueTridentKafkaSpout.java)具有这种特性，同时对kafka节点异常有很好的容错性。OpaqueTridentKafkaSpout在发送一个批次元组的时候，会从上次成功之后的位置开始发送，这样就能够保证元组不会漏发或重发。

基于上面的特点，不透明事务型Spout就不同通过txid来直接判断是否可以跳过状态更新，因为具有相同txid的批次中元组可能发生了变化。

这就需要存储更多的状态信息了，而不仅仅是一个结果和一个txid了，还需要存储前一个结果值。

比如，当前批次的计数是2，需要进行一次状态更新，数据库中的数据如下：

{
  "value": 4,
  "prevValue": 1,
  "txid": 2
}

如果当前的txid是3，与数据库中的不同。在这种情况下，需要将prevValue的值该为value的值，value的值增加2，更新txid为3，最后的结果就是：

{
  "value": 6,
  "prevValue": 4,
  "txid": 3
}

如果当前的txid是2，等于数据库中的txid。因为txid相同，说明上一次txid为2的批次处理失败，但是本次的元组可能与上一次不同了。这个时候，就需要使用本次数据覆盖上次处理结果。也就是说，prevValue值不变，value的值改为prevValue加2，txid不变，最后的结果如下：

{
  "value": 3,
  "prevValue": 1,
  "txid": 2
}

这种方式的可行性依赖于Trident的强顺序性。也就是说，一旦开始处理一个新的批次，就不会重复执行上一个批次。不透明事务型Spout保证了不同批次之间没有重复的情况，也就是每个元组只会在一个批次中处理成功，所以就可以放心的使用前一个值与当前值覆盖已存数据了。

非事务型Spout

非事务型Spout不能为批次提供任何保证。所以可能出现”至多一次”的处理，即在某个批次处理过程中失败了，但是不会在重新处理；也可能提供“至少一次”的处理，即可能会有多个批次分别处理某个元组。也就是没有办法实现“恰好一次”的语义。

不同类型spout和状态总结

下面是不同的spout/状态组合是否支持“恰好一次”处理语义：

不透明事务状态有最强的容错性，但是因为存储txid和两个结果带来更大的开销。事务型状态只需要存储一个状态结果，但是只对事务型Spout有效。非事务型状态要求存储的数据更少，但是不能实现“恰好一次”的处理语义。

所以在选择容错与存储空间中，需要根据具体的需要选择合适的组合。

状态API

根据前面来看，“恰好一次”语义的原理有些复杂，但是作为用户，并不需要了解这些txid对比、多值存储，因为Trident已经在State中封装了所有容错处理逻辑，只需要想下面着用携带码就行：

TridentTopology topology = new TridentTopology();
TridentState wordCounts =
        topology.newStream("spout1", spout)
                .each(new Fields("sentence"), new Split(), new Fields("word"))
                .groupBy(new Fields("word"))
                .persistentAggregate(MemcachedState.opaque(serverLocations), new Count(), new Fields("count"))
                .parallelismHint(6);

所有的不透明事务状态逻辑已经封装在MemcachedState.opaque中，另外，状态更新会自动调整为批次操作，这样可以减少与数据库之间反复交互带来的资源浪费。

基本的State接口只有两个方法：

public interface State {
    void beginCommit(Long txid); // 对于像DRPC流发生的partitionPersist这样的事情，可以是null
    void commit(Long txid);
}

前面已经说过，状态更新开始和结束时都会获取txid。Trident并不关心状态如何操作，使用哪种方式更新，使用哪种方式读取。

假如有一个包含用户地址信息的定制数据库，需要使用Trident与数据库交互，State扩展类中包含对于用户信息的getter和setter方法：

public class LocationDB implements State {
    public void beginCommit(Long txid) {
    }
    public void commit(Long txid) {
    }
    public void setLocation(long userId, String location) {
        // 向数据库设置地址信息
    }
    public String getLocation(long userId) {
        // 从数据库中获取地址信息
    }
}

然后就需要一个StateFactory来创建Trident所需的State对象，LocationDB所需的StateFactory大体结构如下：

public class LocationDBFactory implements StateFactory {
    public State makeState(Map conf, int partitionIndex, int numPartitions) {
        return new LocationDB();
    }
}

Trident提供了用于查询状态源的QueryFunction接口，以及更新状态源的StateUpdater接口。比如，查询LocationDB中用户信息的QueryLocation：

TridentTopology topology = new TridentTopology();
TridentState locations = topology.newStaticState(new LocationDBFactory());
topology.newStream("myspout", spout)
        .stateQuery(locations, new Fields("userid"), new QueryLocation(), new Fields("location"));

QueryLocation的代码如下：

public class QueryLocation extends BaseQueryFunction<LocationDB, String> {
    public List<String> batchRetrieve(LocationDB state, List<TridentTuple> inputs) {
        List<String> ret = new ArrayList();
        for (TridentTuple input : inputs) {
            ret.add(state.getLocation(input.getLong(0)));
        }
        return ret;
    }
    public void execute(TridentTuple tuple, String location, TridentCollector collector) {
        collector.emit(new Values(location));
    }
}

QueryFunction操作分为两步：首先，Trident会将收集到的数据放在一个批次中，发送给batchRetrieve方法。在这个例子中，batchRetrieve方法收到的是一些用户id。batchRetrieve会返回一组与输入元组长度相同的结果。输入元组与输出结果中各个元素是彼此对应的。

从这点来看，上面的LocationDB类并没有发挥Trident批处理优势，所以需要尽心改造：

public class LocationDB implements State {
    public void beginCommit(Long txid) {
    }
    public void commit(Long txid) {
    }
    public void setLocationsBulk(List<Long> userIds, List<String> locations) {
        // set locations in bulk
    }
    public List<String> bulkGetLocations(List<Long> userIds) {
        // get locations in bulk
    }
}

对应的QueryLocation类如下：

public class QueryLocation extends BaseQueryFunction<LocationDB, String> {
    public List<String> batchRetrieve(LocationDB state, List<TridentTuple> inputs) {
        List<Long> userIds = new ArrayList<Long>();
        for (TridentTuple input : inputs) {
            userIds.add(input.getLong(0));
        }
        return state.bulkGetLocations(userIds);
    }
    public void execute(TridentTuple tuple, String location, TridentCollector collector) {
        collector.emit(new Values(location));
    }
}

这段代码大幅减少了数据库操作。

对于更新状态，可以使用StateUpdater接口。比如下面的更新操作：

public class LocationUpdater extends BaseStateUpdater<LocationDB> {
    public void updateState(LocationDB state, List<TridentTuple> tuples, TridentCollector collector) {
        List<Long> ids = new ArrayList<Long>();
        List<String> locations = new ArrayList<String>();
        for (TridentTuple t : tuples) {
            ids.add(t.getLong(0));
            locations.add(t.getString(1));
        }
        state.setLocationsBulk(ids, locations);
    }
}

对应的更新操作拓扑中就可以是这样：

TridentTopology topology = new TridentTopology();
TridentState locations =
        topology.newStream("locations", locationsSpout)
                .partitionPersist(new LocationDBFactory(), new Fields("userid", "location"), new LocationUpdater());

partitionPersist方法会更新状态，StateUpdater接口接收一批元组和状态信息，然后更新状态。上面的LocationUpdater类中仅仅是从元组中抓取用户id和地址信息，然后对状态执行批量处理。然后，partitionPersist会返回一个表示更新状态后的TridentState对象。随后就可以在拓扑的其他地方使用stateQuery方法查询状态。

在StateUpdater的updateState方法中有一个TridentCollector参数，这个对象是可以将发送进来的元组发送到一个新的数据流中。在这个例子中没有用到。如果需要进行比如向数据库更新计数值的后续操作，可以通过TridentState#newValuesStream方法获取新的数据流数据。

persistentAggregate

Trident使用一个名为persistentAggregate的方法更新状态。前面已经出现过，这里再写一遍：

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

partitionPersist是一个接收Trident聚合器作为参数并对状态数据进行更新的方法，persistentAggregate就是构建于partitionPersist上层的一个编程抽象。在这个例子中，通过groupBy返回一个分组数据，Trident需要一个实现MapState接口的对象。分组字段是状态的key，聚合结果是状态的value。MapState接口如下：

public interface MapState<T> extends State {
    List<T> multiGet(List<List<Object>> keys);
    List<T> multiUpdate(List<List<Object>> keys, List<ValueUpdater> updaters);
    void multiPut(List<List<Object>> keys, List<T> vals);
}

如果你需要在未分组的数据流上执行聚合操作时，Trident需要一个实现Snapshottable接口的对象：

public interface Snapshottable<T> extends State {
    T get();
    T update(ValueUpdater updater);
    void set(T o);
}

MemoryMapState 和 MemcachedState 都实现了这些接口.

实现MapState接口

实现MapState接口非常简单，Trident几乎把所有事都做完了。OpaqueMap、TransactionalMap和NonTransactionalMap都分别实现了各自的容错语义。只需要为这些类提供一个用于对不同key/value进行批量获取、批量修改的IBackingMap实现就行。IBackingMap接口如下：

public interface IBackingMap<T> {
    List<T> multiGet(List<List<Object>> keys);
    void multiPut(List<List<Object>> keys, List<T> vals);
}

OpaqueMap会使用OpaqueValue作为vals参数调用multiPut方法；TransactionalMap会使用TransactionalValue作为参数；NonTransactionalMaps会直接把拓扑对象传入。

Trident还提供了CachedMap类来实现key/value的自动LRU缓存操作。

最后，Trident还提供了SnapshottableMap类，该类通过将全局聚合的结果存入一个固定key中的方法将MapState对象转化为Snapshottable对象。

可以参考MemcachedState的实现来了解如何将这些工具结合在一起来提供一个高性能的MapState实现。MemcachedState支持不透明事务、事务和非事务语义。