三位数的域名网站,购买wordpress现有模板,做网站流量怎么解决,惠州市企业网站seo营销工具State
state 可以理解为-- 历史计算结果
有状态计算和无状态计算
无状态计算: 不需要考虑历史数据, 相同的输入,得到相同的输出!如:map, 将每个单词记为1, 进来一个hello, 得到(hello,1),再进来一个hello,得到的还是(hello,1)
有状态计算: 需要考虑历史数据, 相同的输入,可…State
state 可以理解为-- 历史计算结果
有状态计算和无状态计算
无状态计算: 不需要考虑历史数据, 相同的输入,得到相同的输出!如:map, 将每个单词记为1, 进来一个hello, 得到(hello,1),再进来一个hello,得到的还是(hello,1)
有状态计算: 需要考虑历史数据, 相同的输入,可能会得到不同的输出!如:sum/reduce/maxBy, 对单词按照key分组聚合,进来一个(hello,1),得到(hello,1), 再进来一个(hello,1), 得到的结果为(hello,2)
注意: Flink默认已经支持了无状态和有状态计算!
例如WordCount代码:已经做好了状态维护, 输入hello,输出(hello,1),再输入hello,输出(hello,2)
有状态计算和无状态计算的应用场景
无状态计算:数据转换,过滤等操作直接使用无状态的map/filter即可有状态计算:需要做聚合/比较的操作得使用有状态的sum/reduce/maxBy/minBy....
有状态中的状态的分类
有状态的计算是流处理框架要实现的重要功能因为稍复杂的流处理场景都需要记录状态State然后在新流入数据的基础上不断更新状态。下面的几个场景都需要使用流处理的状态功能
数据流中的数据有重复想对重复数据去重需要记录哪些数据已经流入过应用当新数据流入时根据已流入过的数据来判断去重。检查输入流是否符合某个特定的模式需要将之前流入的元素以状态的形式缓存下来。比如判断一个温度传感器数据流中的温度是否在持续上升。对一个时间窗口内的数据进行聚合分析分析一个小时内某项指标的75分位或99分位的数值。
其实窗口本身就是状态他不是立即出结果而是将数据都保存起来达到触发条件才计算。
一个状态更新和获取的流程如下图所示一个算子子任务接收输入流获取对应的状态根据新的计算结果更新状态。一个简单的例子是对一个时间窗口内输入流的某个整数字段求和那么当算子子任务接收到新元素时会获取已经存储在状态中的数值然后将当前输入加到状态上并将状态数据更新。 以wordcout为例说明上图的流程 状态类型
Flink有两种基本类型的状态托管状态Managed State和原生状态Raw State。
两者的区别Managed State是由Flink管理的Flink帮忙存储、恢复和优化Raw State是开发者自己管理的需要自己序列化。
具体区别有
从状态管理的方式上来说Managed State由Flink Runtime托管状态是自动存储、自动恢复的Flink在存储管理和持久化上做了一些优化。当横向伸缩或者说修改Flink应用的并行度时状态也能自动重新分布到多个并行实例上。Raw State是用户自定义的状态。
从状态的数据结构上来说Managed State支持了一系列常见的数据结构如ValueState、ListState、MapState等。Raw State只支持字节任何上层数据结构需要序列化为字节数组。使用时需要用户自己序列化以非常底层的字节数组形式存储Flink并不知道存储的是什么样的数据结构。
从具体使用场景来说绝大多数的算子都可以通过继承Rich函数类或其他提供好的接口类在里面使用Managed State。Raw State是在已有算子和Managed State不够用时用户自定义算子时使用。
对Managed State继续细分它又有两种类型Keyed State和Operator State。
Flink状态 - 托管状态- KeyedState ( 在keyBy之后可以使用状态 )- ValueState (存储一个值)- ListState (存储多个值)- MapState (存储key-value) - OperatorState ( 没有keyBy的情况下也可以使用 ) [不用]- 原生状态 (不用)
Keyed State 键控状态
Flink 为每个键值维护一个状态实例并将具有相同键的所有数据都分区到同一个算子任务中这个任务会维护和处理这个key对应的状态。当任务处理一条数据时它会自动将状态的访问范围限定为当前数据的key。因此具有相同key的所有数据都会访问相同的状态。
需要注意的是键控状态只能在 KeyedStream 上进行使用可以通过 stream.keyBy(...) 来得到 KeyedStream 。 Flink 提供了以下数据格式来管理和存储键控状态 (Keyed State)
· ValueState存储单值类型的状态。可以使用 update(T) 进行更新并通过 T value() 进行检索。
· ListState存储列表类型的状态。可以使用 add(T) 或 addAll(List) 添加元素并通过 get() 获得整个列表。
· ReducingState用于存储经过 ReduceFunction 计算后的结果使用 add(T) 增加元素。
· AggregatingState用于存储经过 AggregatingState 计算后的结果使用 add(IN) 添加元素。
· FoldingState已被标识为废弃会在未来版本中移除官方推荐使用 AggregatingState 代替。
· MapState维护 Map 类型的状态。
代码演示-Managed State-Keyed State
//nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/datastream/fault-tolerance/state/
案例1:
使用KeyedState中的ValueState获取数据中的最大值获取每个key的最大值(实际中直接使用maxBy即可)
也就是我们自己使用KeyState中的ValueState来模拟实现maxBy
代码实现:
package com.bigdata.state;import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class _01_KeyedStateDemo {public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//2. source-加载数据DataStreamTuple2String, Long tupleDS env.fromElements(Tuple2.of(北京, 1L),Tuple2.of(上海, 2L),Tuple2.of(北京, 6L),Tuple2.of(上海, 8L),Tuple2.of(北京, 3L),Tuple2.of(上海, 4L),Tuple2.of(北京, 7L));//2. source-加载数据tupleDS.keyBy(new KeySelectorTuple2String, Long, String() {Overridepublic String getKey(Tuple2String, Long value) throws Exception {return value.f0;}}).map(new RichMapFunctionTuple2String, Long, Tuple2String,Long() {// 借助状态这个API实现ValueStateLong maxValueState null;Overridepublic void open(Configuration parameters) throws Exception {// 就是对ValueState初始化ValueStateDescriptorLong stateDescriptor new ValueStateDescriptorLong(valueState,Long.class);maxValueState getRuntimeContext().getState(stateDescriptor);}Overridepublic Tuple2String, Long map(Tuple2String, Long value) throws Exception {Long val value.f1;if(maxValueState.value() null){maxValueState.update(val);}else{if(maxValueState.value() val){maxValueState.update(val);}}return Tuple2.of(value.f0,maxValueState.value());}}).print();//.maxBy(1).print();//3. transformation-数据处理转换//4. sink-数据输出//5. execute-执行env.execute();}
}
案例2:
如果一个人的体温超过阈值38度超过3次及以上则输出: 姓名 [温度1,温度2,温度3] 姓名,温度输入 输出张三,37张三,38张三,39张三,35张三,40张三,41 张三,[39,40,41]张三,40 张三,[39,40,41,40]
package com.bigdata.state;import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.planner.expressions.In;
import org.apache.flink.util.Collector;import java.util.ArrayList;public class _02_KeyedStateDemo2 {// 如果一个人的体温超过阈值38度超过3次及以上则输出: 姓名 [温度1,温度2,温度3]public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//2. source-加载数据DataStreamSourceString dataStreamSource env.socketTextStream(localhost, 8889);//3. transformation-数据处理转换 zs,37dataStreamSource.map(new MapFunctionString, Tuple2String,Integer() {Overridepublic Tuple2String, Integer map(String value) throws Exception {String[] arr value.split(,);return Tuple2.of(arr[0],Integer.valueOf(arr[1]));}}).keyBy(new KeySelectorTuple2String, Integer, String() {Overridepublic String getKey(Tuple2String, Integer value) throws Exception {return value.f0;}}).flatMap(new RichFlatMapFunctionTuple2String, Integer, Tuple2String, ArrayListInteger() {ValueStateInteger valueState null;ListStateInteger listState null;Overridepublic void open(Configuration parameters) throws Exception {ValueStateDescriptorInteger stateDescriptor new ValueStateDescriptorInteger(numState,Integer.class);valueState getRuntimeContext().getState(stateDescriptor);ListStateDescriptorInteger listStateDescriptor new ListStateDescriptor(listState, Integer.class);listState getRuntimeContext().getListState(listStateDescriptor);}Overridepublic void flatMap(Tuple2String, Integer value, CollectorTuple2String, ArrayListInteger out) throws Exception {Integer tiwen value.f1;if(tiwen 38){valueState.update(valueState.value()null?1:(valueState.value()1));listState.add(tiwen);}if(valueState.value()!null valueState.value() 3){ArrayListInteger list new ArrayList();IterableInteger iterable listState.get();for (Integer tiwenwen : iterable) {list.add(tiwenwen);}out.collect(Tuple2.of(value.f0,list));}}}).print();//4. sink-数据输出//5. execute-执行env.execute();}
}
