共计 8784 个字符，预计需要花费 22 分钟才能阅读完成。
序
本文主要研究一下 storm trident 的 state
StateType
storm-2.0.0/storm-client/src/jvm/org/apache/storm/trident/state/StateType.java
public enum StateType {
NON_TRANSACTIONAL,
TRANSACTIONAL,
OPAQUE
}
StateType 有三种类型，NON_TRANSACTIONAL 非事务性，TRANSACTIONAL 事务性，OPAQUE 不透明事务
对应的 spout 也有三类，non-transactional、transactional 以及 opaque transactional
State
storm-2.0.0/storm-client/src/jvm/org/apache/storm/trident/state/State.java
/**
* There’s 3 different kinds of state:
*
* 1. non-transactional: ignores commits, updates are permanent. no rollback. a cassandra incrementing state would be like this 2.
* repeat-transactional: idempotent as long as all batches for a txid are identical 3. opaque-transactional: the most general kind of state.
* updates are always done based on the previous version of the value if the current commit = latest stored commit Idempotent even if the
* batch for a txid can change.
*
* repeat transactional is idempotent for transactional spouts opaque transactional is idempotent for opaque or transactional spouts
*
* Trident should log warnings when state is idempotent but updates will not be idempotent because of spout
*/
// retrieving is encapsulated in Retrieval interface
public interface State {
void beginCommit(Long txid); // can be null for things like partitionPersist occuring off a DRPC stream
void commit(Long txid);
}
non-transactional，忽略 commits，updates 是持久的，没有 rollback，cassandra 的 incrementing state 属于这个类型；at-most 或者 at-least once 语义
repeat-transactional，简称 transactional，要求不管是否 replayed，同一个 batch 的 txid 始终相同，而且里头的 tuple 也不变，一个 tuple 只属于一个 batch，各个 batch 之间不会重叠；对于 state 更新来说，replay 遇到相同的 txid，即可跳过；在数据库需要较少的 state，但是容错性较差，保证 exactly once 语义
opaque-transactional，简称 opaque，是用的比较多的一类，它的容错性比 transactional 强，它不要求一个 tuple 始终在同一个 batch/txid，也就是说允许一个 tuple 在这个 batch 处理失败，但是在其他 batch 中处理成功，但是它可以保证每个 tuple 只在某一个 batch 中 exactly 成功处理一次；OpaqueTridentKafkaSpout 就是这个类型的实现，它能容忍 kafka 节点丢失的错误；对于 state 更新来说，replay 遇到相同的 txid，则需要基于 prevValue 使用当前的值覆盖掉；在数据库需要更多空间来存储 state，但是容错性好，保证 exactly once 语义
MapState
storm-2.0.0/storm-client/src/jvm/org/apache/storm/trident/state/map/MapState.java
public interface MapState<T> extends ReadOnlyMapState<T> {
List<T> multiUpdate(List<List<Object>> keys, List<ValueUpdater> updaters);
void multiPut(List<List<Object>> keys, List<T> vals);
}
MapState 继承了 ReadOnlyMapState 接口，而 ReadOnlyMapState 则继承了 State 接口
这里主要举 MapState 的几个实现类分析一下
NonTransactionalMap
storm-2.0.0/storm-client/src/jvm/org/apache/storm/trident/state/map/NonTransactionalMap.java
public class NonTransactionalMap<T> implements MapState<T> {
IBackingMap<T> _backing;
protected NonTransactionalMap(IBackingMap<T> backing) {
_backing = backing;
}
public static <T> MapState<T> build(IBackingMap<T> backing) {
return new NonTransactionalMap<T>(backing);
}
@Override
public List<T> multiGet(List<List<Object>> keys) {
return _backing.multiGet(keys);
}
@Override
public List<T> multiUpdate(List<List<Object>> keys, List<ValueUpdater> updaters) {
List<T> curr = _backing.multiGet(keys);
List<T> ret = new ArrayList<T>(curr.size());
for (int i = 0; i < curr.size(); i++) {
T currVal = curr.get(i);
ValueUpdater<T> updater = updaters.get(i);
ret.add(updater.update(currVal));
}
_backing.multiPut(keys, ret);
return ret;
}
@Override
public void multiPut(List<List<Object>> keys, List<T> vals) {
_backing.multiPut(keys, vals);
}
@Override
public void beginCommit(Long txid) {
}
@Override
public void commit(Long txid) {
}
}
NonTransactionalMap 包装了 IBackingMap，beginCommit 及 commit 方法都不做任何操作
multiUpdate 方法构造 List<T> ret，然后使用 IBackingMap 的 multiPut 来实现
TransactionalMap
storm-2.0.0/storm-client/src/jvm/org/apache/storm/trident/state/map/TransactionalMap.java
public class TransactionalMap<T> implements MapState<T> {
CachedBatchReadsMap<TransactionalValue> _backing;
Long _currTx;
protected TransactionalMap(IBackingMap<TransactionalValue> backing) {
_backing = new CachedBatchReadsMap(backing);
}
public static <T> MapState<T> build(IBackingMap<TransactionalValue> backing) {
return new TransactionalMap<T>(backing);
}
@Override
public List<T> multiGet(List<List<Object>> keys) {
List<CachedBatchReadsMap.RetVal<TransactionalValue>> vals = _backing.multiGet(keys);
List<T> ret = new ArrayList<T>(vals.size());
for (CachedBatchReadsMap.RetVal<TransactionalValue> retval : vals) {
TransactionalValue v = retval.val;
if (v != null) {
ret.add((T) v.getVal());
} else {
ret.add(null);
}
}
return ret;
}
@Override
public List<T> multiUpdate(List<List<Object>> keys, List<ValueUpdater> updaters) {
List<CachedBatchReadsMap.RetVal<TransactionalValue>> curr = _backing.multiGet(keys);
List<TransactionalValue> newVals = new ArrayList<TransactionalValue>(curr.size());
List<List<Object>> newKeys = new ArrayList();
List<T> ret = new ArrayList<T>();
for (int i = 0; i < curr.size(); i++) {
CachedBatchReadsMap.RetVal<TransactionalValue> retval = curr.get(i);
TransactionalValue<T> val = retval.val;
ValueUpdater<T> updater = updaters.get(i);
TransactionalValue<T> newVal;
boolean changed = false;
if (val == null) {
newVal = new TransactionalValue<T>(_currTx, updater.update(null));
changed = true;
} else {
if (_currTx != null && _currTx.equals(val.getTxid()) && !retval.cached) {
newVal = val;
} else {
newVal = new TransactionalValue<T>(_currTx, updater.update(val.getVal()));
changed = true;
}
}
ret.add(newVal.getVal());
if (changed) {
newVals.add(newVal);
newKeys.add(keys.get(i));
}
}
if (!newKeys.isEmpty()) {
_backing.multiPut(newKeys, newVals);
}
return ret;
}
@Override
public void multiPut(List<List<Object>> keys, List<T> vals) {
List<TransactionalValue> newVals = new ArrayList<TransactionalValue>(vals.size());
for (T val : vals) {
newVals.add(new TransactionalValue<T>(_currTx, val));
}
_backing.multiPut(keys, newVals);
}
@Override
public void beginCommit(Long txid) {
_currTx = txid;
_backing.reset();
}
@Override
public void commit(Long txid) {
_currTx = null;
_backing.reset();
}
}
TransactionalMap 采取的是 CachedBatchReadsMap<TransactionalValue>，这里泛型使用的是 TransactionalValue，beginCommit 会设置当前的 txid，重置_backing，commit 的时候会重置 txid，然后重置_backing
multiUpdate 方法中判断如果_currTx 已经存在值，且该值!retval.cached(即不是本次事务中 multiPut 进去的)，那么不会更新该值 (skip the update)，使用 newVal = val
multiPut 方法构造批量的 TransactionalValue，然后使用 CachedBatchReadsMap.multiPut(List<List<Object>> keys, List<T> vals) 方法，该方法更新值之后会更新到缓存
OpaqueMap
storm-2.0.0/storm-client/src/jvm/org/apache/storm/trident/state/map/OpaqueMap.java
public class OpaqueMap<T> implements MapState<T> {
CachedBatchReadsMap<OpaqueValue> _backing;
Long _currTx;
protected OpaqueMap(IBackingMap<OpaqueValue> backing) {
_backing = new CachedBatchReadsMap(backing);
}
public static <T> MapState<T> build(IBackingMap<OpaqueValue> backing) {
return new OpaqueMap<T>(backing);
}
@Override
public List<T> multiGet(List<List<Object>> keys) {
List<CachedBatchReadsMap.RetVal<OpaqueValue>> curr = _backing.multiGet(keys);
List<T> ret = new ArrayList<T>(curr.size());
for (CachedBatchReadsMap.RetVal<OpaqueValue> retval : curr) {
OpaqueValue val = retval.val;
if (val != null) {
if (retval.cached) {
ret.add((T) val.getCurr());
} else {
ret.add((T) val.get(_currTx));
}
} else {
ret.add(null);
}
}
return ret;
}
@Override
public List<T> multiUpdate(List<List<Object>> keys, List<ValueUpdater> updaters) {
List<CachedBatchReadsMap.RetVal<OpaqueValue>> curr = _backing.multiGet(keys);
List<OpaqueValue> newVals = new ArrayList<OpaqueValue>(curr.size());
List<T> ret = new ArrayList<T>();
for (int i = 0; i < curr.size(); i++) {
CachedBatchReadsMap.RetVal<OpaqueValue> retval = curr.get(i);
OpaqueValue<T> val = retval.val;
ValueUpdater<T> updater = updaters.get(i);
T prev;
if (val == null) {
prev = null;
} else {
if (retval.cached) {
prev = val.getCurr();
} else {
prev = val.get(_currTx);
}
}
T newVal = updater.update(prev);
ret.add(newVal);
OpaqueValue<T> newOpaqueVal;
if (val == null) {
newOpaqueVal = new OpaqueValue<T>(_currTx, newVal);
} else {
newOpaqueVal = val.update(_currTx, newVal);
}
newVals.add(newOpaqueVal);
}
_backing.multiPut(keys, newVals);
return ret;
}
@Override
public void multiPut(List<List<Object>> keys, List<T> vals) {
List<ValueUpdater> updaters = new ArrayList<ValueUpdater>(vals.size());
for (T val : vals) {
updaters.add(new ReplaceUpdater<T>(val));
}
multiUpdate(keys, updaters);
}
@Override
public void beginCommit(Long txid) {
_currTx = txid;
_backing.reset();
}
@Override
public void commit(Long txid) {
_currTx = null;
_backing.reset();
}
static class ReplaceUpdater<T> implements ValueUpdater<T> {
T _t;
public ReplaceUpdater(T t) {
_t = t;
}
@Override
public T update(Object stored) {
return _t;
}
}
}
OpaqueMap 采取的是 CachedBatchReadsMap<OpaqueValue>，这里泛型使用的是 OpaqueValue，beginCommit 会设置当前的 txid，重置_backing，commit 的时候会重置 txid，然后重置_backing
与 TransactionalMap 的不同，这里在 multiPut 的时候，使用的是 ReplaceUpdater，然后调用 multiUpdate 强制覆盖
multiUpdate 方法与 TransactionalMap 的不同，它是基于 prev 值来进行 update 的，算出 newVal
小结
trident 严格按 batch 的顺序更新 state，比如 txid 为 3 的 batch 必须在 txid 为 2 的 batch 处理完之后才能处理
state 分三种类型，分别是 non-transactional、transactional、opaque transactional，对应的 spout 也是这三种类型
non-transactional 无法保证 exactly once，它可能是 at-least once 或者 at-most once；其 state 计算参考 NonTransactionalMap，对于 beginCommit 及 commit 操作都无处理
transactional 类型能够保证 exactly once，但是要求比较严格，要同一个 batch 的 txid 及 tuple 在 replayed 的时候仍然保持一致，因此容错性差一点，但是它的 state 计算相对简单，参考 TransactionalMap，遇到同一个 txid 的值，skip 掉即可
opaque transactional 类型也能够保证 exactly once，它允许一个 tuple 处理失败之后，出现在其他 batch 中处理，因而容错性好，但是 state 计算要多存储 prev 值，参考 OpaqueMap，遇到同一个 txid 的值，使用 prev 值跟当前值进行覆盖
trident 将保证 exactly once 的 state 的计算都封装好了，使用的时候，在 persistentAggregate 传入相应的 StateFactory 即可，支持多种 StateType 的 factory 可以选择使用 StateType 属性，通过传入不同的参数构造不同 transactional 的 state；也可以通过实现 StateFactory 自定义实现 state factory，另外也可以通过继承 BaseQueryFunction 来自定义 stateQuery 查询，自定义更新的话，可以继承 BaseStateUpdater，然后通过 partitionPersist 传入
doc
Trident Tutorial
Trident State
