Flink四大基石之State

devtools/2024/11/30 8:41:00/

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。

java">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

代码实现:

java">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-加载数据DataStream<Tuple2<String, 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 KeySelector<Tuple2<String, Long>, String>() {@Overridepublic String getKey(Tuple2<String, Long> value) throws Exception {return value.f0;}}).map(new RichMapFunction<Tuple2<String, Long>, Tuple2<String,Long>>() {// 借助状态这个API实现ValueState<Long> maxValueState= null;@Overridepublic void open(Configuration parameters) throws Exception {// 就是对ValueState初始化ValueStateDescriptor<Long> stateDescriptor = new ValueStateDescriptor<Long>("valueState",Long.class);maxValueState = getRuntimeContext().getState(stateDescriptor);}@Overridepublic Tuple2<String, Long> map(Tuple2<String, 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]

 

java">姓名,温度输入                      输出张三,37张三,38张三,39张三,35张三,40张三,41               张三,[39,40,41]张三,40               张三,[39,40,41,40]
java">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-加载数据DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 8889);//3. transformation-数据处理转换   zs,37dataStreamSource.map(new MapFunction<String, Tuple2<String,Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {String[] arr = value.split(",");return Tuple2.of(arr[0],Integer.valueOf(arr[1]));}}).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {@Overridepublic String getKey(Tuple2<String, Integer> value) throws Exception {return value.f0;}}).flatMap(new RichFlatMapFunction<Tuple2<String, Integer>, Tuple2<String, ArrayList<Integer>>>() {ValueState<Integer> valueState = null;ListState<Integer> listState = null;@Overridepublic void open(Configuration parameters) throws Exception {ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<Integer>("numState",Integer.class);valueState = getRuntimeContext().getState(stateDescriptor);ListStateDescriptor<Integer> listStateDescriptor = new ListStateDescriptor<>("listState", Integer.class);listState = getRuntimeContext().getListState(listStateDescriptor);}@Overridepublic void flatMap(Tuple2<String, Integer> value, Collector<Tuple2<String, ArrayList<Integer>>> 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){ArrayList<Integer> list = new ArrayList<>();Iterable<Integer> iterable = listState.get();for (Integer tiwenwen : iterable) {list.add(tiwenwen);}out.collect(Tuple2.of(value.f0,list));}}}).print();//4. sink-数据输出//5. execute-执行env.execute();}
}

 

 

 


http://www.ppmy.cn/devtools/138142.html

相关文章

HTTP（网络）

HTTP（网络）

目录 1.Http的基本代码 1.1 HttpServer.hpp 1.2 简单测试一下 1.3 用telnet测试一下 1.4 用浏览器访问 1.5 返回相应的过程&#xff08;网页版本&#xff09;​编辑 1.5.1 再次用浏览器访问 1.6 返回相应的过程&#xff08;文件版本&#xff09; 1.6.1网页 1.6.2 测试 …
阅读更多...
亚马逊开发视频人工智能模型，The Information 报道

亚马逊开发视频人工智能模型，The Information 报道

根据《The Information》周三的报道&#xff0c;电子商务巨头亚马逊&#xff08;AMZN&#xff09;已开发出一种新的生成式人工智能&#xff08;AI&#xff09;&#xff0c;不仅能处理文本&#xff0c;还能处理图片和视频&#xff0c;从而减少对人工智能初创公司Anthropic的依赖…
阅读更多...
鸿蒙技术分享：Navigation页面管理-鸿蒙@fw/router框架源码解析（二）

鸿蒙技术分享：Navigation页面管理-鸿蒙@fw/router框架源码解析（二）

theme: smartblue 本文是系列文章&#xff0c;其他文章见&#xff1a;鸿蒙fw/router框架源码解析&#xff08;一&#xff09;-Router页面管理 鸿蒙fw/router框架源码解析 介绍 fw/router是在HarmonyOS鸿蒙系统中开发应用所使用的开源模块化路由框架。该路由框架基于模块化开…
阅读更多...
【第十课】Rust并发编程（一）

【第十课】Rust并发编程（一）

目录 前言 Fork和Join 前言 本节会介绍Rust中的并发编程&#xff0c;并发编程在编程中是提升cpu使用率的一大利器&#xff0c;通过多线程技术提升效率&#xff0c;Rust的并发和其他编程语言的并发不同的地方在于&#xff0c;Rust号称无畏并发。更重要的一点是安全。Rust中所有…
阅读更多...
# issue 6 网络编程基础

# issue 6 网络编程基础

一、网络的物理结构和光纤千兆网络 首先&#xff0c;我们需要知道网络的物理结构——数据是如何从一台机器传输到另外一台机器的 这个过程是非常重要的。现在很多人做软件开发&#xff0c;只会软件角度&#xff0c;这导致讲软件原理头头是道&#xff0c;但是连数据线都不会接&a…
阅读更多...
基于OpenCV视觉库让机械手根据视觉判断物体有无和分类抓取的例程

基于OpenCV视觉库让机械手根据视觉判断物体有无和分类抓取的例程

项目实例&#xff0c;在一个无人封闭的隔绝场景中&#xff0c;根据视觉判断物件的有无&#xff0c;通过机械手 进行物件分类提取&#xff0c;并且返回状态结果&#xff1b; 实际的场景是有一个类似采血的固件支架盘&#xff0c;上面很多采血管&#xff0c;采血管帽颜色可能不同…
阅读更多...
深入探索Flax：一个用于构建神经网络的灵活和高效库

深入探索Flax：一个用于构建神经网络的灵活和高效库

深入探索Flax&#xff1a;一个用于构建神经网络的灵活和高效库 在深度学习领域&#xff0c;TensorFlow 和 PyTorch 作为主流的框架&#xff0c;已被广泛使用。不过&#xff0c;Flax 作为一个较新的库&#xff0c;近年来得到了越来越多的关注。Flax 是一个由Google Research团队…
阅读更多...
租赁小程序|租赁系统搭建|租赁系统需求

租赁小程序|租赁系统搭建|租赁系统需求

随着信息技术的高速发展&#xff0c;租赁行业逐渐向智能化、便捷化方向迈进。一款优秀的租赁小程序&#xff0c;旨在为用户提供一站式的租赁服务体验&#xff0c;同时帮助租赁企业优化管理流程&#xff0c;提高业务效率。 一、用户需求精准把握 在开发任何软件产品时&#xff0…
阅读更多...
最新文章

Copyright @ 2022~2023