聊聊storm trident的state

本文主要研究一下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 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,beginCommit会设置当前的txid,重置_backing,commit的时候会重置txid,然后重置_backing
  • multiUpdate方法中判断如果_currTx已经存在值,且该值!retval.cached(即不是本次事务中multiPut进去的),那么不会更新该值(skip the update),使用newVal = val
  • multiPut方法构造批量的TransactionalValue,然后使用CachedBatchReadsMap.multiPut(List> keys, List 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,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

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏大闲人柴毛毛

Java8新特性——StreamAPI(一)

1. 流的基本概念 1.1 什么是流? 流是Java8引入的全新概念,它用来处理集合中的数据,暂且可以把它理解为一种高级集合。 众所周知,集合操作非常麻烦,若...

3559
来自专栏Python爬虫与算法进阶

学点算法之栈的学习与应用

在学习栈前,脑海中对这个词只有一个印象:客栈 栈是什么 栈(有时称为“后进先出栈”)是一个项的有序集合,其中添加移除新项总发生在同一端。 这段话初学者是懵逼的...

3093
来自专栏我杨某人的青春满是悔恨

深入理解 weak-strong dance

这时handler持有 Block 对象,而 Block 对象虽然捕获了weakSelf,延长了weakSelf这个局部变量的生命周期,但weakSelf是附有...

924
来自专栏码代码的陈同学

JVM 栈和栈帧

对于没有深度递归的函数来说,无需担心上篇文章中的算法。当知道正在处理数据集有限时,我会使用这种简单的基本递归形式。由于你并不知道在应用程序中会处理多少数据,因此...

7109
来自专栏JAVA高级架构

JAVA基础面试总结

1.00 什么时候使用基于接口编程? 基于接口编程、Fascade层等等抽象封装都是有开发和维护的代价的,是否使用归根结底还是要看团队人员的分工情况, 技术方...

3208
来自专栏SDNLAB

《P4语言规范》Header & Instances详解

一、前言 为深入研究P4语言相关规范及运行操作使用,本系列文章根据P4.org网站给出的《The P4 Language Specification v1.0....

3318
来自专栏Python爱好者

Python高效编程(五)

1053
来自专栏维C果糖

Guava 指南 之「前置条件」

前置条件 Guava 提供了很多用于进行前置条件检查的工具,我们强烈建议静态导入这些方法。 每个方法都用三种形式: 没有额外的参数。抛出的任何异常都没有错误信息...

2027
来自专栏拭心的安卓进阶之路

Java 集合源码解析(2):ListIterator

今天心情和股票一样红,还是学学 ListIterator 吧! ListIterator ? 根据官方文档介绍, ListIterator 有以下功能: 允许...

2119
来自专栏hbbliyong

一个小程序引发的思考

   既然是一个小程序引发的思考,那么我们就先看看这个小程序,看看他有何神奇之处: namespace ConsoleApplication1 { cl...

2414

扫码关注云+社区