前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
圈层
工具
发布
首页
学习
活动
专区
圈层
工具
MCP广场
社区首页 >专栏 >Apache-Flink深度解析-State

Apache-Flink深度解析-State

作者头像
王知无-import_bigdata
修改于 2024-03-12 01:54:38
修改于 2024-03-12 01:54:38
1.4K00
代码可运行
举报
运行总次数:0
代码可运行

转载自:https://dwz.cn/xrMCqbk5

摘要:

实际问题 在流计算场景中,数据会源源不断的流入Apache Flink系统,每条数据进入Apache Flink系统都会触发计算。如果我们想进行一个Count聚合计算,那么每次触发计算是将历史上所有流入的数据重新新计算一次,还是每次计算都是在上一次计算结果之上进行增量计算呢?答案是肯定的,Apache Flink是基于上一次的计算结果进行增量计算的。

实际问题

在流计算场景中,数据会源源不断的流入Apache Flink系统,每条数据进入Apache Flink系统都会触发计算。如果我们想进行一个Count聚合计算,那么每次触发计算是将历史上所有流入的数据重新新计算一次,还是每次计算都是在上一次计算结果之上进行增量计算呢?答案是肯定的,Apache Flink是基于上一次的计算结果进行增量计算的。那么问题来了: "上一次的计算结果保存在哪里,保存在内存可以吗?",答案是否定的,如果保存在内存,在由于网络,硬件等原因造成某个计算节点失败的情况下,上一次计算结果会丢失,在节点恢复的时候,就需要将历史上所有数据(可能十几天,上百天的数据)重新计算一次,所以为了避免这种灾难性的问题发生,Apache Flink 会利用State存储计算结果。本篇将会为大家介绍Apache Flink State的相关内容。

什么是State

这个问题似乎有些"弱智"?不管问题的答案是否显而易见,但我还是想简单说一下在Apache Flink里面什么是State?State是指流计算过程中计算节点的中间计算结果或元数据属性,比如 在aggregation过程中要在state中记录中间聚合结果,比如 Apache Kafka 作为数据源时候,我们也要记录已经读取记录的offset,这些State数据在计算过程中会进行持久化(插入或更新)。所以Apache Flink中的State就是与时间相关的,Apache Flink任务的内部数据(计算数据和元数据属性)的快照。

为什么需要State

与批计算相比,State是流计算特有的,批计算没有failover机制,要么成功,要么重新计算。流计算在 大多数场景 下是增量计算,数据逐条处理(大多数场景),每次计算是在上一次计算结果之上进行处理的,这样的机制势必要将上一次的计算结果进行存储(生产模式要持久化),另外由于 机器,网络,脏数据等原因导致的程序错误,在重启job时候需要从成功的检查点(checkpoint,后面篇章会专门介绍)进行state的恢复。增量计算,Failover这些机制都需要state的支撑。

State 实现

Apache Flink内部有四种state的存储实现,具体如下:

  • 基于内存的HeapStateBackend - 在debug模式使用,不 建议在生产模式下应用;
  • 基于HDFS的FsStateBackend - 分布式文件持久化,每次读写都产生网络IO,整体性能不佳;
  • 基于RocksDB的RocksDBStateBackend - 本地文件+异步HDFS持久化;
  • 还有一个是基于Niagara(Alibaba内部实现)NiagaraStateBackend - 分布式持久化- 在Alibaba生产环境应用;

State 持久化逻辑

Apache Flink版本选择用RocksDB+HDFS的方式进行State的存储,State存储分两个阶段,首先本地存储到RocksDB,然后异步的同步到远程的HDFS。 这样而设计既消除了HeapStateBackend的局限(内存大小,机器坏掉丢失等),也减少了纯分布式存储的网络IO开销。

State 分类

Apache Flink 内部按照算子和数据分组角度将State划分为如下两类:

  • KeyedState - 这里面的key是我们在SQL语句中对应的GroupBy/PartitioneBy里面的字段,key的值就是groupby/PartitionBy字段组成的Row的字节数组,每一个key都有一个属于自己的State,key与key之间的State是不可见的;
  • OperatorState - Apache Flink内部的Source Connector的实现中就会用OperatorState来记录source数据读取的offset。

State 扩容重新分配

Apache Flink是一个大规模并行分布式系统,允许大规模的有状态流处理。 为了可伸缩性,Apache Flink作业在逻辑上被分解成operator graph,并且每个operator的执行被物理地分解成多个并行运算符实例。 从概念上讲,Apache Flink中的每个并行运算符实例都是一个独立的任务,可以在自己的机器上调度到网络连接的其他机器运行。

Apache Flink的DAG图中只有边相连的节点有网络通信,也就是整个DAG在垂直方向有网络IO,在水平方向如下图的stateful节点之间没有网络通信,这种模型也保证了每个operator实例维护一份自己的state,并且保存在本地磁盘(远程异步同步)。通过这种设计,任务的所有状态数据都是本地的,并且状态访问不需要任务之间的网络通信。 避免这种流量对于像Apache Flink这样的大规模并行分布式系统的可扩展性至关重要。

如上我们知道Apache Flink中State有OperatorState和KeyedState,那么在进行扩容时候(增加并发)State如何分配呢?比如:外部Source有5个partition,在Apache Flink上面由Srouce的1个并发扩容到2个并发,中间Stateful Operation 节点由2个并发并扩容的3个并发,如下图所示:

在Apache Flink中对不同类型的State有不同的扩容方法,接下来我们分别介绍。

OperatorState对扩容的处理

我们选取Apache Flink中某个具体Connector实现实例进行介绍,以MetaQ为例,MetaQ以topic方式订阅数据,每个topic会有N>0个分区,以上图为例,加上我们订阅的MetaQ的topic有5个分区,那么当我们source由1个并发调整为2个并发时候,State是怎么恢复的呢? state 恢复的方式与Source中OperatorState的存储结构有必然关系,我们先看MetaQSource的实现是如何存储State的。首先MetaQSource 实现了ListCheckpointed<T extends Serializable>,其中的T是Tuple2<InputSplit,Long>,我们在看ListCheckpointed接口的内部定义如下:

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
public interface ListCheckpointed<T extends Serializable>; {
List<T> snapshotState(long var1, long var3) throws Exception;

void restoreState(List&lt;T&gt; var1) throws Exception;
}

我们发现 snapshotState方法的返回值是一个List<T>,T是Tuple2<InputSplit,Long>,也就是snapshotState方法返回List<Tuple2<InputSplit,Long>>,这个类型说明state的存储是一个包含partiton和offset信息的列表,InputSplit代表一个分区,Long代表当前partition读取的offset。InputSplit有一个方法如下:

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
public interface InputSplit extends Serializable {
int getSplitNumber();
}

也就是说,InputSplit我们可以理解为是一个Partition索引,有了这个数据结构我们在看看上面图所示的case是如何工作的?当Source的并行度是1的时候,所有打partition数据都在同一个线程中读取,所有partition的state也在同一个state中维护,State存储信息格式如下:

如果我们现在将并发调整为2,那么我们5个分区的State将会在2个独立的任务(线程)中进行维护,在内部实现中我们有如下算法进行分配每个Task所处理和维护partition的State信息,如下:

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
List<Integer> assignedPartitions = new LinkedList<>();
for (int i = 0; i < partitions; i++) {
if (i % consumerCount == consumerIndex) {
assignedPartitions.add(i);
}
}

这个求mod的算法,决定了每个并发所处理和维护partition的State信息,针对我们当前的case具体的存储情况如下:

那么到现在我们发现上面扩容后State得以很好的分配得益于OperatorState采用了List<T>的数据结构的设计。另外大家注意一个问题,相信大家已经发现上面分配partition的算法有一个限制,那就是Source的扩容(并发数)是否可以超过Source物理存储的partition数量呢?答案是否定的,不能。目前Apache Flink的做法是提前报错,即使不报错也是资源的浪费,因为超过partition数量的并发永远分配不到待管理的partition。

KeyedState对扩容的处理

对于KeyedState最容易想到的是hash(key) mod parallelism(operator) 方式分配state,就和OperatorState一样,这种分配方式大多数情况是恢复的state不是本地已有的state,需要一次网络拷贝,这种效率比较低,OperatorState采用这种简单的方式进行处理是因为OperatorState的state一般都比较小,网络拉取的成本很小,对于KeyedState往往很大,我们会有更好的选择,在Apache Flink中采用的是Key-Groups方式进行分配。

什么是Key-Groups

Key-Groups 是Apache Flink中对keyed state按照key进行分组的方式,每个key-group中会包含N>0个key,一个key-group是State分配的原子单位。在Apache Flink中关于Key-Group的对象是 KeyGroupRange, 如下:

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
public class KeyGroupRange implements KeyGroupsList, Serializable {
...
...
private final int startKeyGroup;
private final int endKeyGroup;
...
...
}

KeyGroupRange两个重要的属性就是 startKeyGroup和endKeyGroup,定义了startKeyGroup和endKeyGroup属性后Operator上面的Key-Group的个数也就确定了。

什么决定Key-Groups的个数

key-group的数量在job启动前必须是确定的且运行中不能改变。由于key-group是state分配的原子单位,而每个operator并行实例至少包含一个key-group,因此operator的最大并行度不能超过设定的key-group的个数,那么在Apache Flink的内部实现上key-group的数量就是最大并行度的值。

GroupRange.of(0, maxParallelism)如何决定key属于哪个Key-Group 确定好GroupRange之后,如何决定每个Key属于哪个Key-Group呢?我们采取的是取mod的方式,在KeyGroupRangeAssignment中的assignToKeyGroup方法会将key划分到指定的key-group中,如下:

如上实现我们了解到分配Key到指定的key-group的逻辑是利用key的hashCode和maxParallelism进行取余操作来分配的。如下图当parallelism=2,maxParallelism=10的情况下流上key与key-group的对应关系如下图所示:

如上图key(a)的hashCode是97,与最大并发10取余后是7,被分配到了KG-7中,流上每个event都会分配到KG-0至KG-9其中一个Key-Group中。 每个Operator实例如何获取Key-Groups 了解了Key-Groups概念和如何分配每个Key到指定的Key-Groups之后,我们看看如何计算每个Operator实例所处理的Key-Groups。 在KeyGroupRangeAssignment的computeKeyGroupRangeForOperatorIndex方法描述了分配算法:

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
public static KeyGroupRange computeKeyGroupRangeForOperatorIndex(
int maxParallelism,
int parallelism,
int operatorIndex) {
GroupRange splitRange = GroupRange.of(0, maxParallelism).getSplitRange(parallelism, operatorIndex);
int startGroup = splitRange.getStartGroup();
int endGroup = splitRange.getEndGroup();
return new KeyGroupRange(startGroup, endGroup - 1);
}

public GroupRange getSplitRange(int numSplits, int splitIndex) {
...
final int numGroupsPerSplit = getNumGroups() / numSplits;
final int numFatSplits = getNumGroups() % numSplits;

int startGroupForThisSplit;
int endGroupForThisSplit;
if (splitIndex &lt; numFatSplits) {
startGroupForThisSplit = getStartGroup() + splitIndex * (numGroupsPerSplit + 1);
endGroupForThisSplit = startGroupForThisSplit + numGroupsPerSplit + 1;
} else {
startGroupForThisSplit = getStartGroup() + splitIndex * numGroupsPerSplit + numFatSplits;
endGroupForThisSplit = startGroupForThisSplit + numGroupsPerSplit;
}
if (startGroupForThisSplit &gt;= endGroupForThisSplit) {
return GroupRange.emptyGroupRange();
} else {
return new GroupRange(startGroupForThisSplit, endGroupForThisSplit);
}
}

上面代码的核心逻辑是先计算每个Operator实例至少分配的Key-Group个数,将不能整除的部分N个,平均分给前N个实例。最终每个Operator实例管理的Key-Groups会在GroupRange中表示,本质是一个区间值;下面我们就上图的case,说明一下如何进行分配以及扩容后如何重新分配。 假设上面的Stateful Operation节点的最大并行度maxParallelism的值是10,也就是我们一共有10个Key-Group,当我们并发是2的时候和并发是3的时候分配的情况如下图:

如上算法我们发现在进行扩容时候,大部分state还是落到本地的,如Task0只有KG-4被分出去,其他的还是保持在本地。同时我们也发现,一个job如果修改了maxParallelism的值那么会直接影响到Key-Groups的数量和key的分配,也会打乱所有的Key-Group的分配,目前在Apache Flink系统中统一将maxParallelism的默认值调整到4096,最大程度的避免无法扩容的情况发生。

本篇简单介绍了Apache Flink中State的概念,并重点介绍了OperatorState和KeyedState在扩容时候的处理方式。Apache Flink State是支撑Apache Flink中failover,增量计算,Window等重要机制和功能的核心设施。后续介绍failover,增量计算,Window等相关篇章中也会涉及State的利用,当涉及到本篇没有覆盖的内容时候再补充介绍。

本文系转载,前往查看

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

本文系转载,前往查看

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

评论
登录后参与评论
暂无评论
推荐阅读
编辑精选文章
换一批
数据之舞——在MySQL中奏响数据库操作的乐章
在数字的世界里,MySQL像是一艘承载数据的船,带领我们穿越信息的海洋。每一条SQL语句,都是我们与数据库之间的语言桥梁。我们借助这些命令,去查询隐藏在深处的答案,去更新和重组混乱的片段。在这条旅途中,逻辑与技巧是我们的罗盘,指引我们穿越复杂的表结构与关系。
suye
2025/05/29
860
数据之舞——在MySQL中奏响数据库操作的乐章
Linux——MySQL基础
也就是说,mysql是一套提供数据存储服务的网络程序。 数据库一般指的是在磁盘或者内存中存储的特定结构组织的数据——将来在磁盘中存储数据的一套特定方案。 数据库服务就是mysqld。 为什么有数据库 虽然一般文件确实提供了数据存储的功能,但是站在用户角度上,文件并没有提供非常好的数据管理能力。 数据库的本质:对数据内容存取的一套解决方案,你给我数据内容,我直接给你结果。
有礼貌的灰绅士
2025/05/10
1010
Linux——MySQL基础
初识MySQL · 库的操作
由前文可得,MySQL是目前主流的数据库,mysql是客户端,mysqld是一种网络服务,mysqld是一种数据库服务,而对于数据库来说,是一种存储数据、组织数据的一种手段,那么我们肯定就是要学数据库是如何操作的?对于数据库我们应该如何增删查改?删除了如何备份等问题。
_lazy
2024/10/23
1030
初识MySQL · 库的操作
全面解析MySQL数据库与表的操作:从创建到备份恢复,掌握所有技巧
当然我们也可以创建我们想要指定创建的相关的字符集以及校验规则的数据库 创建一个叫db2的数据库,设置的字符集为utf8
用户11286421
2025/03/16
1650
全面解析MySQL数据库与表的操作:从创建到备份恢复,掌握所有技巧
MySQL库操作、表操作
实例操作: 创建一个database1的数据库: 指令:create database database1;
南桥
2024/07/26
2150
MySQL库操作、表操作
【MySQL】库的操作
MySQL是一个数据库管理系统,在其中我们可以创建许多的数据库,数据库中中又可以存储许多的表。 如下图所示:
Yui_
2025/04/18
820
【MySQL】库的操作
【MySql】数据库的增删改查
当我们创建数据没有指定字符集和校验规则时,系统使用默认字符集:utf8,校验规则:utf_general_ci;
平凡的人1
2023/10/15
5100
【MySql】数据库的增删改查
【MySQL】库和表的操作
数据库校验集  --  支持数据库进行字段比较使用的编码,本质也是一种读取数据库中数据采用的编码格式。
青衫哥
2023/10/17
2670
【MySQL】库和表的操作
【MySQL篇】高效管理MySQL:数据库操作与优化的终极指南
本篇博客将带你一步步探索MySQL数据库的操作技巧,涵盖从数据库创建、表结构设计到数据查询、更新及优化的各个方面。无论你是初学者还是有一定经验的开发者,本文都将为你提供实用的技巧和深入的见解,让你能够在日常工作中游刃有余地管理和操作MySQL数据库。准备好一起开启MySQL的探索之旅了吗?让我们从基础开始,逐步深入,掌握真正的MySQL操作技巧!
熬夜学编程的小王
2025/04/05
2040
【MySQL篇】高效管理MySQL:数据库操作与优化的终极指南
【MySQL】数据库操作指南:轻松驾驭数据世界
我们已经查看查看完数据库,接下来创建自己没有的数据库,比如创建一个名为test1和test2的数据库:
喜欢做梦
2025/02/18
1230
【MySQL】数据库操作指南:轻松驾驭数据世界
【MySQL】从零开始:掌握MySQL数据库的核心概念
风中的云彩
2025/03/18
1180
【MySQL】从零开始:掌握MySQL数据库的核心概念
【MySQL】库和表的操作
在创建数据库的时候后面可以跟上选项,采用我们指定的字符集和校验规则,但是如果不加选项就会使用配置文件中默认的字符集和校验规则。
用户11305458
2025/03/12
1330
【MySQL】库和表的操作
库的操作(MySQL)
说明:当我们创建数据库没有指定字符集和校验规则时,系统使用默认字符集:utf8,校验规则 是:utf8_ general_ ci
ljw695
2024/11/21
1090
【MySQL基础篇】二、数据库的操作
​ 接下来学的数据库的操作以及表的结构操作,都是属于我们之前讲的 DDL 也就是数据定义语言,它负责的是对结构的管理,而不是对数据内容的管理!
利刃大大
2025/05/21
920
【MySQL基础篇】二、数据库的操作
MySQL库的操作
创建一个数据库Test1,校验规则使用utf8_ general_ ci[不区分大小写]:
二肥是只大懒蓝猫
2023/10/13
2250
MySQL库的操作
MySQL数据库介绍及基础操作
2.SQL语言的分类: DDL数据定义语言,用来维护存储数据的结构 代表指令: create, drop, alter DML数据操纵语言,用来对数据进行操作 代表指令: insert,delete,update DML中又单独分了一个DQL,数据查询语言,代表指令: select DCL数据控制语言,主要负责权限管理和事务
用户11305962
2024/10/09
1320
MySQL数据库介绍及基础操作
【MySQL】数据库基础&&库/表的操作&&数据类型详解
为解决上述问题,专家们设计出更加利于管理数据的东西-数据库,能更加有效的管理数据,数据库的水平是衡量一个程序员水平的重要指标
用户10925563
2025/02/16
1240
【MySQL】数据库基础&&库/表的操作&&数据类型详解
【MySQL】003. MySQL操作库
💦创建数据库的本质就是在/var/lib/mysql/路径下创建一个目录 删除数据库
_孙同学
2025/04/17
710
【MySQL】003. MySQL操作库
MySQL库的操作『增删改查 ‖ 编码问题 ‖ 备份与恢复』
[ ] 中的关键字可以选择不加,IF NOT EXISTS 的意思是如果不存在才创建 数据库
北 海
2023/11/15
2220
MySQL库的操作『增删改查 ‖ 编码问题 ‖ 备份与恢复』
从零开始学Mysql - 字符集和编码(上)
从零开始学Mysql - 字符集和编码(上) 前言 上一节我们系统的阐述了关于系统配置的相关细节内容,而这一节我们需要了解关于字符集和编码的内容,字符集和编码的规则其实也算是入门mysql经常遇到的
阿东
2021/12/08
8360
从零开始学Mysql - 字符集和编码(上)
相关推荐
数据之舞——在MySQL中奏响数据库操作的乐章
更多 >
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档
本文部分代码块支持一键运行,欢迎体验
本文部分代码块支持一键运行,欢迎体验