如何制作网站页面,wordpress hover,福州专业做网站公司,自己做的网站突然打不开一、什么是状态
无状态计算的例子#xff1a; 例如一个加法算子#xff0c;第一次输入235那么以后我多次数据23的时候得到的结果都是5。得出的结论就是#xff0c;相同的输入都会得到相同的结果#xff0c;与次数无关。 有状态计算的例子#xff1a; 访问量的统计#x…一、什么是状态
无状态计算的例子 例如一个加法算子第一次输入235那么以后我多次数据23的时候得到的结果都是5。得出的结论就是相同的输入都会得到相同的结果与次数无关。 有状态计算的例子 访问量的统计我们都知道Nginx的访问日志一个请求一条日志基于此我们就可以统计访问量。如下/api/a这个url第一此访问的时候返回的结果就是 count1但当第二次访问的时候返回的结果变成了2。为什么Flink知道之前已经处理过一次 hello world这就是state发挥作用了这里是被称为keyed state存储了之前需要统计的数据keyby接口的调用会创建keyed stream对key进行划分这是使用keyed state的前提。得出的结论就是相同的输入得到不同的结果与次数有关。这就是有状态的数据。
什么场景下会大量使用到这种状态数据啦简单举几个例子 【1】去重的需求中比如说我们只想知道这100个同事都属于那几个部门的等等。 【2】窗口计算已进入未触发的数据。比如我们一分钟统计一次1-2之间的1.5这个时候的数据对于2来说就是一个有状态的数据因为2的结果与1.5有关。 【3】机器学习/深度学习训练的模型及参数。这对于机器学习的同学深入感触。比如第一次输入hello机器会给我一个反馈那么下次会基于这个反馈做进一步的学习处理。那么上一步的结果对于我而言就是一种有状态的输入。 【4】访问历史数据需要与昨日进行对比。昨日的数据对于今日而言也属于一种状态。你品你细品。
为什么要管理状态用内存不香吗首先流失作业是有它的标准的不是什么东西随随便便就说自己这个是流失处理。首先7*24小时运行高可靠你内存不行吧你的容量总有用完的时候吧。其次数据不丢失不重恰好计算一次你内存要实现需要备份和恢复你还总伴随着小部分数据的丢失吧。最后数据实时产生不延迟你内存不够横向扩展时你需要延迟吧。
理想的状态管理就是下面描述的样子Flink也都帮我们实现了。
二、状态的类型
Managed State Raw State
Managed StateRaw State状态管理方式Flink Runtime 管理 —自动存储自动恢复 —内存管理上有优化用户自己管理Flink不知道你在State中存储的数据结构的 —要自己实例化状态数据结构已知的数据结构 —value,list,map…字节数据 —byte[]推荐使用场景大多数情况下均可使用自定义 Operator 时可以使用当Managed State 不够时使用
Managed Stated 分为 Keyed Stated和Operator State 【1】Keyed Stated 只能用于keyBy生成的KeyedStream上的算子。每一个key对应一个State一个Operator实例处理多个Key访问相应的多个State。相同Key会在相同的实例中处理。整个过程如果没有keyBy操作它是没有KeyedStream的而Keyed Stated只能应用在KeyedStream 上。
并发改变 State随着Key在实例间迁移。例如实例A中之前处理KeyA与KeyB后面我扩展了实例B那么 实例A就只需要处理KeyAKeyB就交给 实例B进行处理。安装状态进行分离可以理解为分布式。
通过 RuntimeContext 访问说明Operator是一个Rich Function否则是拿不到RuntimeContext。
支持的数据结构 ValueState、ListState、ReducingState、AggregatingState、MapState
【2】Operator State 可以用于所有的算子常用于source上例如FlinkKafkaConsumer。一个Operator实例对应一个State所以一个Operator中会处理多个key可以理解为集群。
并发改变 Operator State没有key并发改变的时候就需要重新分配。内置了两种方案均匀分配和合并后每个得到全量。
访问方式 实现CheckpointedFunction或ListCheckpointed接口。
支持的数据结构 ListState
三、Keyed State 使用示例
什么是 keyed state 对于keyed state有两个特点 【1】只能应用于KeyedStream 的函数与操作中例如Keyed UDF, window state 【2】keyed state是已经分区 / 划分好的每一个 key 只能属于某一个 keyed state 对于如何理解已经分区的概念我们需要看一下keyby的语义大家可以看到下图左边有三个并发右边也是三个并发左边的词进来之后通过keyby会进行相应的分发。例如对于hello wordhello这个词通过hash运算永远只会到右下方并发的task上面去。
什么是 operator state 【1】又称为non-keyed state每一个operator state都仅与一个operator的实例绑定。 【2】常见的operator state是source state例如记录当前source的offset再看一段使用operator state的word count代码
这里的fromElements会调用FromElementsFunction的类其中就使用了类型为list state的operator state。如下几种Keyed State之间的依赖关系都是state的子类。它们的访问方式和数据结构都有一定的区别。
状态数据类型访问接口备注ValueState单个值[update(T) 修改/T value 获取]例如 WordCount 用 word 做 keystate就是单个的数值。这个单个也可以是字符串、对象等都有可能。访问方式只有上面两种。MapStateMapput(UK key, UV value) putAll(MapUK,UV map) remove(UK key) boolean contains(UK key) UV get(UK key) IterableMap.Entry entries() IterableMap.Entry iterator() Iterable keys() Iterable values()能够操作具体的对象的keyListStateListadd/ addAll(List) update(List) Iterable get()ReducingState单个值add/ addAll(List) update(List) T get()与 List 是同一个父类这个add是直接将数据更新进了 Reducing的结果里面。举个例子例如我们统计1分钟的结果list是先将数据添加到list中等到1分钟的时候全来出来统计。而 Reducing是来一条就统计一条结果。好处是节省内存。AggregatingState单个值add(IN)/OUT get()与 List 是同一个父类与Reducing的不同是Reducing输入和输出的类型都是相同的。而Aggregating 是可以不同的。例如我要计算一个平局值Reducing是算好返回而Aggregating会返回总和和个数。
举个ValueState的案例
final StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();
//获取数据流
DataStreamEvent events env.addSource(source);DataStreamAlert alerts events// 生成 keyedStata 通过 sourceAddress.keyBy(Event::sourceAddress)// StateMachineMapper 状态机.flatMap(new StateMachineMapper());//我么看下状态机怎么写 实现 RichFlatMapFunction
SuppressWarnings(serial)
static class StateMachineMapper extends RichFlatMapFunctionEvent, Alert {private ValueStateLeaderLatch.State currentState;Overridepublic void open(Configuration conf) {// 获取一个 valueStatecurrentState getRuntimeContext().getState(new ValueStateDescriptor(state, State.class));}//来一条数据处理一条Overridepublic void flatMap(Event evt, CollectorAlert out) throws Exception {// 获取 valueState state currentState.value();if (state null) {state State.Initial;//State 是本地的变量}// 把事件对状态的影响加上去得到一个状态State nextState state.transition(evt.type());//判断状态是否合法if (nextState State.InvalidTransition) {//扔出去out.collect(new Alert(evt.sourceAddress(), state, evt.type()));}//是否不能继续转化了例如取消的订单else if (nextState.isTerminal()) {// 从 state 中清楚掉currentState.clear();}else {// 修改状态currentState.update(nextState);}}
}四、CheckPoint 与 state 的关系
Checkpoint是从source触发到下游所有节点完成的一次全局操作。下图可以有一个对Checkpoint的直观感受红框里面可以看到一共触发了 569K次Checkpoint然后全部都成功完成没有fail的。
**state 其实就是 Checkpoint 所做的主要持久化备份的主要数据**看下图的具体数据统计其state也就9kb大小 。
五、状态如何保存和恢复
Checkpoint定时制作分布式快照对程序的状态进行备份。发生故障时将整个作业的Task都回滚到最后一次成功Checkpoint中的状态然后从保存的点继续处理。
必要条件 数据源支持重发如果不重发丢失的消息就真的丢了
一致性语义 恰好一次如果p相同单线程多个线程时可能有的算子对其已经计算了一次了有的没有就需要注意至少一次。
// 获取运行环境
final StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();
//状态数据
//两个checkpoint 触发间隔设置1S越频繁追的数据就越少io消耗也越大
env.enableCheckpointing(1000);
//EXACTLY_ONCE语义说明 Checkpoint是要对替的这样消息不会重复也不会对丢。
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//两个checkpoint 最少等待500ms 例如第一个checkpoint做了700ms按理300ms后就要做下一个checkpoint。但是它们之间的等待时间300ms500ms 此时就会延长200ms减少checkpoint过于频繁影响业务。
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//checkpoint多久超时如果这个checkpoint在1分钟内还没做完那就失败了
env.getCheckpointConfig().setCheckpointTimeout(60000);
//同时最多有多少个checkpoint进行
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//当重新分配并发度拆分task时是否保存checkpoint。如果不保存就需要使用savepoint来保存数据放到外部的介质中。
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);Checkpoint vs Savepoint
CheckpointSavepoint触发管理方式由Flink自动触发并管理由用户手动触发并管理主要用途在 Task 发生异常时快速恢复例如网络抖动导致的超时异常有计划的进行备份使作业能停止后再恢复例如修改代码、调整并发。特点轻量、自动从故障中服务、在作业停止后默认清除持久、以标准格式存储允许代码或配置发生变化、手动触发 savepoint 恢复。
可选的状态存储方式 【1】MemoryStateBackend构造方法
MemoryStateBackend(int maxStateSize, boolean asynchronousSnapshots)存储方式 StateTaskManager内存。CheckpointJobManager内存。 容量限制 单个State maxStateSize默认5M。maxStateSize akka.framesize默认10M。总大小不超过JobManager内存。 推荐使用场景 本地测试几乎无状态的作业比如ETL/JobManager不容易挂或影响不大的情况。不推荐在生产场景使用。
【2】FsStateBackend 构造方法
FsStateBackend(URL checkpointDataUri, boolean asynchronousSnapshots)存储方式 StateTaskManager内存。Checkpoint外部文件系统本地或HDFS。 容量限制 单个TaskManager上State总量不超过它的内存。总大小不超过配置的文件系统容量会定期清理。 推荐使用场景 常规使用状态的作业例如分钟级窗口聚合、join。需要开启HA的作业。可以在生产环境使用。
【3】RocksDBStateBackend 构造方法
RocksDBStateBackend(URL checkpointDataUri, boolean enableIncrementalCheckpointing)存储方式 StateTaskManager上的KV数据库实际使用内存磁盘。Checkpoint外部文件系统本地或HDFS。 容量限制 单个TaskManager上State总量不超过它的内存磁盘单个key 最大2G。总大小不超过配置的文件系统容量。 推荐使用场景 超大状态的作业例如天级窗口聚合。需要开启HA的作业。对状态读写性能要求比较高的作业。可以在生产环境使用。
