当前位置: 首页 > news >正文

Flink四大基石之State

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-加载数据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]

 

姓名,温度输入                      输出张三,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-加载数据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();}
}

 

 

 

相关文章:

Flink四大基石之State

State state 可以理解为-- 历史计算结果 有状态计算和无状态计算 无状态计算: 不需要考虑历史数据, 相同的输入,得到相同的输出!如:map, 将每个单词记为1, 进来一个hello, 得到(hello,1),再进来一个hello,得到的还是(hello,1) 有状态计算: 需要考虑历史数据, 相同的输入,可…...

Spacy小笔记:zh_core_web_trf、zh_core_web_lg、zh_core_web_md 和 zh_core_web_sm区别

Spacy小笔记 最近频繁用到spacy&#xff0c;就小记一下。 2024.11.29 zh_core_web_trf、zh_core_web_lg、zh_core_web_md 和 zh_core_web_sm区别 首先&#xff0c;它们都是预训练的中文模型&#xff1a; zh_core_web_trf:395M 架构: 基于 Transformer 架构&#xff08;bert…...

第六届智能控制、测量与信号处理国际学术会议 (ICMSP 2024)

重要信息 2024年11月29日-12月1日 中国陕西西安石油大学雁塔校区 大会官网&#xff1a;www.icmsp.net 大会简介 第六届智能控制、测量与信号处理国际学术会议&#xff08;ICMSP 2024&#xff09;由西安石油大学、中海油田服务股份有限公司、浙江水利水电学院与中国石油装备…...

docker服务容器化

docker服务容器化 1 引言2 多个容器间网络联通2.1 单独创建关联2.2 创建时关联 3 服务搭建3.1 镜像清单3.2 容器创建 4 联合实战4.2 flink_sql之kafka到starrocks4.2 flink_sql之mysql到starrocks 5 文献借鉴 1 引言 ​ 利用docker可以很效率地搭建服务&#xff0c;本文在win1…...

【QT】控件8

1.QDial 通过调节旋钮位置来控制窗口的不透明度&#xff1a; void Widget::on_dial_valueChanged(int value) {qDebug()<<value;this->setWindowOpacity((double)value/100); }效果演示&#xff1a; 2.Date/Time Edit 计算两个日期的差值 ui界面设计 计算按钮按下…...

漫谈推理谬误——错误因果

相关文章 漫谈推理谬误——错误假设-CSDN博客文章浏览阅读736次&#xff0c;点赞22次&#xff0c;收藏3次。在日常生活中&#xff0c;我们会面临各种逻辑推理&#xff0c;有些看起来一目了然&#xff0c;有些非常的科学严谨&#xff0c;但也有很多似是而非&#xff0c;隐藏了陷…...

【数据结构】队列实现剖析:掌握队列的底层实现

在计算机科学中&#xff0c;**队列&#xff08;Queue&#xff09;**是一种常见的数据结构&#xff0c;它遵循先进先出&#xff08;FIFO&#xff0c;First In First Out&#xff09;的原则。队列的应用非常广泛&#xff0c;例如任务调度、资源管理、进程通信等。本篇文章旨在为计…...

【C++】IO库(二):文件输入输出

8.2 文件输入输出 头文件 fstream 定义了三个类型来之支持文件IO&#xff0c;分别是&#xff1a; ifstream&#xff1a;从一个给定文件读取数据&#xff1b;ofstream&#xff1a;向一个给定文件写入数据&#xff1b;fstream&#xff1a;读写给定文件。 在 C 当中&#xff0c…...

105.【C语言】数据结构之二叉树求总节点和第K层节点的个数

目录 1.求二叉树总的节点的个数 1.容易想到的方法 代码 缺陷 思考:能否在TreeSize函数内定义静态变量解决size的问题呢? 其他写法 运行结果 2.最好的方法:分而治之 代码 运行结果 2.求二叉树第K层节点的个数 错误代码 运行结果 修正 运行结果 其他写法 1.求二…...

力扣637. 二叉树的层平均值

给定一个非空二叉树的根节点 root , 以数组的形式返回每一层节点的平均值。与实际答案相差 10-5 以内的答案可以被接受。 提示&#xff1a; 树中节点数量在 [1, 104] 范围内-231 < Node.val < 231 - 1 代码&#xff1a; /*** Definition for a binary tree node.* stru…...

【前端】Next.js 服务器端渲染(SSR)与客户端渲染(CSR)的最佳实践

关于Next.js 服务器端渲染&#xff08;SSR&#xff09;与客户端渲染&#xff08;CSR&#xff09;的实践内容方面&#xff0c;我们按下面几点进行阐述。 1. 原理 服务器端渲染 (SSR): 在服务器上生成完整的HTML页面&#xff0c;然后发送给客户端。这使得用户在首次访问时能够…...

路径规划之启发式算法之一:A-Star(A*)算法

A*算法是一种启发式搜索算法&#xff0c;常用于解决路径规划问题。 一、A*算法的定义与原理 A*算法是一种用于在图形或网格中查找最短路径的算法。它在搜索过程中综合考虑了每个节点的实际距离&#xff08;g值&#xff09;和预估距离&#xff08;h值&#xff09;&#xff0c;以…...

Android复习代码1-4章

public class RudioButton extends AppCompatActivity {Overrideprotected void onCreate(Nullable Bundle savedInstanceState) {super.onCreate(savedInstanceState);setContentView(R.layout.activity_rudio_button);// 找到RadioGroup和TextView的实例RadioGroup radioGrou…...

【问题】webdriver.Chrome()设置参数executable_path报不存在

场景1: 标红报错unresolved reference executable_path 场景2: 执行报错TypeError: __init__() got an unexpected keyword argument executable_path 原因&#xff1a; 上述两种场景是因为selenium4开始不再支持某些初始化参数。比如executable_path 解决&#xff1a; 方案…...

win10系统安装docker-desktop

1、开启Hyper-v ———————————————— Hyper-V 是微软提供的一种虚拟化技术&#xff0c;它允许你在同一台物理计算机上运行多个独立的操作系统实例。这种技术主要用于开发、测试、以及服务器虚拟化等领域。 —————————————————————— &#…...

小程序-基于java+SpringBoot+Vue的乡村研学旅行平台设计与实现

项目运行 1.运行环境&#xff1a;最好是java jdk 1.8&#xff0c;我们在这个平台上运行的。其他版本理论上也可以。 2.IDE环境&#xff1a;IDEA&#xff0c;Eclipse,Myeclipse都可以。推荐IDEA; 3.tomcat环境&#xff1a;Tomcat 7.x,8.x,9.x版本均可 4.硬件环境&#xff1a…...

组件A底部栏(position: fixed )事件使用$emit更新内容失败bug解决

今天遇到一个很离奇的bug&#xff0c;记录一下 问题&#xff1a;在组件内底部栏使用$emit触发按钮事件但打印出来的值是初始化的值&#xff0c;更新的值被重置导致更新失败 原因&#xff1a;组件内底部使用了 position: fixed; 固定&#xff0c; 导致组件内插槽 this 与 保存按…...

数据结构——排序第三幕(深究快排(非递归实现)、快排的优化、内省排序,排序总结)超详细!!!!

文章目录 前言一、非递归实现快排二、快排的优化版本三、内省排序四、排序算法复杂度以及稳定性的分析总结 前言 继上一篇博客基于递归的方式学习了快速排序和归并排序 今天我们来深究快速排序&#xff0c;使用栈的数据结构非递归实现快排&#xff0c;优化快排&#xff08;三路…...

C++的类功能整合

1. 类的基本概念 类是面向对象编程的核心&#xff0c;它封装了数据和操作数据的函数。 #include <iostream> using namespace std;class MyClass { public:int publicData;void publicFunction() {cout << "Public function" << endl;}private:i…...

《String类》

目录 一、定义与概述 二、创建字符串对象 2.1 直接赋值 2.2 使用构造函数 三、字符串的不可变性 四、常用方法 4.1 String对象的比较 4.1.1 比较是否引用同一个对象 4.1.2 boolean equals(Object anObject)方法&#xff1a;按照字典序比较 4.1.3 int compareTo(Strin…...

【WiFi帧结构】

文章目录 帧结构MAC头部管理帧 帧结构 Wi-Fi的帧分为三部分组成&#xff1a;MAC头部frame bodyFCS&#xff0c;其中MAC是固定格式的&#xff0c;frame body是可变长度。 MAC头部有frame control&#xff0c;duration&#xff0c;address1&#xff0c;address2&#xff0c;addre…...

SciencePlots——绘制论文中的图片

文章目录 安装一、风格二、1 资源 安装 # 安装最新版 pip install githttps://github.com/garrettj403/SciencePlots.git# 安装稳定版 pip install SciencePlots一、风格 简单好用的深度学习论文绘图专用工具包–Science Plot 二、 1 资源 论文绘图神器来了&#xff1a;一行…...

【HarmonyOS 5.0】DevEco Testing:鸿蒙应用质量保障的终极武器

——全方位测试解决方案与代码实战 一、工具定位与核心能力 DevEco Testing是HarmonyOS官方推出的​​一体化测试平台​​&#xff0c;覆盖应用全生命周期测试需求&#xff0c;主要提供五大核心能力&#xff1a; ​​测试类型​​​​检测目标​​​​关键指标​​功能体验基…...

大数据零基础学习day1之环境准备和大数据初步理解

学习大数据会使用到多台Linux服务器。 一、环境准备 1、VMware 基于VMware构建Linux虚拟机 是大数据从业者或者IT从业者的必备技能之一也是成本低廉的方案 所以VMware虚拟机方案是必须要学习的。 &#xff08;1&#xff09;设置网关 打开VMware虚拟机&#xff0c;点击编辑…...

现代密码学 | 椭圆曲线密码学—附py代码

Elliptic Curve Cryptography 椭圆曲线密码学&#xff08;ECC&#xff09;是一种基于有限域上椭圆曲线数学特性的公钥加密技术。其核心原理涉及椭圆曲线的代数性质、离散对数问题以及有限域上的运算。 椭圆曲线密码学是多种数字签名算法的基础&#xff0c;例如椭圆曲线数字签…...

【JavaWeb】Docker项目部署

引言 之前学习了Linux操作系统的常见命令&#xff0c;在Linux上安装软件&#xff0c;以及如何在Linux上部署一个单体项目&#xff0c;大多数同学都会有相同的感受&#xff0c;那就是麻烦。 核心体现在三点&#xff1a; 命令太多了&#xff0c;记不住 软件安装包名字复杂&…...

让回归模型不再被异常值“带跑偏“,MSE和Cauchy损失函数在噪声数据环境下的实战对比

在机器学习的回归分析中&#xff0c;损失函数的选择对模型性能具有决定性影响。均方误差&#xff08;MSE&#xff09;作为经典的损失函数&#xff0c;在处理干净数据时表现优异&#xff0c;但在面对包含异常值的噪声数据时&#xff0c;其对大误差的二次惩罚机制往往导致模型参数…...

Go 语言并发编程基础:无缓冲与有缓冲通道

在上一章节中&#xff0c;我们了解了 Channel 的基本用法。本章将重点分析 Go 中通道的两种类型 —— 无缓冲通道与有缓冲通道&#xff0c;它们在并发编程中各具特点和应用场景。 一、通道的基本分类 类型定义形式特点无缓冲通道make(chan T)发送和接收都必须准备好&#xff0…...

LLMs 系列实操科普(1)

写在前面&#xff1a; 本期内容我们继续 Andrej Karpathy 的《How I use LLMs》讲座内容&#xff0c;原视频时长 ~130 分钟&#xff0c;以实操演示主流的一些 LLMs 的使用&#xff0c;由于涉及到实操&#xff0c;实际上并不适合以文字整理&#xff0c;但还是决定尽量整理一份笔…...

接口自动化测试:HttpRunner基础

相关文档 HttpRunner V3.x中文文档 HttpRunner 用户指南 使用HttpRunner 3.x实现接口自动化测试 HttpRunner介绍 HttpRunner 是一个开源的 API 测试工具&#xff0c;支持 HTTP(S)/HTTP2/WebSocket/RPC 等网络协议&#xff0c;涵盖接口测试、性能测试、数字体验监测等测试类型…...