Flink中的状态管理

小灰灰 2023-01-17 03:28 268阅读 0赞

一、状态定义

  • 由一个任务维护,并且用来计算某个结果的所有数据,都属于这个任务的状态
  • 可以认为状态就是一个本地变量,可以被任务的业务逻辑访问
  • Flink会进行状态管理,包括状态一致性、故障处理以及高效存储和访问,以便开发人员可以专注于应用程序的逻辑
    在这里插入图片描述

二、状态管理

在Flink中,状态始终与特定算子相关联。为了使运行时的Flink了解算子的状态,算子需要预先注册其状态

总的来说,有两种类型的状态:

  • 算子状态(Operator State):算子状态的作用范围限定为算子任务
  • 键控状态(Keyed State):根据输入数据流中定义的键(key)来维护和访问

2.1 算子状态(Operator State)

算子状态的作用范围限定为算子任务,由同一并行任务所处理的所有数据都可以访问到相同的状态;状态对于同一子任务而言是共享的;算子状态不能被相同或不同算子的另一个子任务访问
在这里插入图片描述

2.1.1 算子状态数据结构

  • 列表状态(List state):将状态表示为一组数据的列表
  • 联合列表状态(Union list state):也将状态表示为数据的列表。它与常规列表状态的区别在于,在发生故障时,或者从保存点(savepoint)启动应用程序时如何恢复
  • 广播状态(Broadcast state):如果一个算子有多项任务,而它的每项任务状态又都相同,那么这种特殊情况最适合应用广播状态

2.1.2 示例代码

  1. public class StateTest1_OperatorState {
  2. public static void main(String[] args) throws Exception {
  3. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4. env.setParallelism(1);
  5. // socket文本流
  6. DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
  7. // 转换成SensorReading类型
  8. // java8 中的lamda表达式
  9. DataStream<SensorReading> dataStream = inputStream.map(line -> {
  10. String[] fields = line.split(",");
  11. return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
  12. });
  13. //定义一个有状态的map操作,统计当前分区数据个数
  14. SingleOutputStreamOperator<Integer> resultStream = dataStream.map(new MyCountMapper());
  15. resultStream.print();
  16. env.execute();
  17. }
  18. //自定义MapFunction
  19. public static class MyCountMapper implements MapFunction<SensorReading, Integer>, ListCheckpointed<Integer> {
  20. //定义一个本地变量,作为算子状态
  21. private Integer count = 0;
  22. @Override
  23. public Integer map(SensorReading value) throws Exception {
  24. count++;
  25. return count;
  26. }
  27. @Override
  28. public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
  29. return Collections.singletonList(count);
  30. }
  31. @Override
  32. public void restoreState(List<Integer> state) throws Exception {
  33. for(Integer num : state){
  34. count += num;
  35. }
  36. }
  37. }
  38. }

2.2 键控状态(Keyed State)

键控状态是根据输入数据流中定义的键(key)来维护和访问的;Flink为每个key维护一个状态实例,并将具有相同键的所有数据,都分区到同一个算子任务中,这个任务会维护和处理这个key对应的状态;当任务处理一条数据时,它会自动将状态的访问范围限定为当前数据的key

2.2.1 键控状态数据结构

  • 值状态(Value state):将状态表示为单个的值
  • 列表状态(List state):将状态表示为一组数据的列表
  • 映射状态(Map state):将状态表示为一组 Key-Value 对
  • 聚合状态(Reducing state & Aggregating state):将状态表示为一个用于聚合操作的列表

2.2.2 键控状态的使用

  • 声明一个键控状态

    myValueState = getRuntimeContext().getState(

    1. new ValueStateDescriptor<Integer>(
    2. "my-value",
    3. Integer.class));
  • 读取状态

    Integer myValue = myValueState.value();

  • 对状态赋值

    myValueState.update(10);

值状态的示例代码:

  1. public class StateTest2_KeyedState {
  2. public static void main(String[] args) throws Exception {
  3. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4. env.setParallelism(1);
  5. // socket文本流
  6. DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
  7. // 转换成SensorReading类型
  8. // java8 中的lamda表达式
  9. DataStream<SensorReading> dataStream = inputStream.map(line -> {
  10. String[] fields = line.split(",");
  11. return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
  12. });
  13. //定义一个有状态的map操作,统计当前sensor数据个数
  14. SingleOutputStreamOperator<Integer> resultStream = dataStream.keyBy("id")
  15. .map(new MyKeyCountMapper());
  16. resultStream.print();
  17. env.execute();
  18. }
  19. //自定义实现一个RichMapFunction
  20. public static class MyKeyCountMapper extends RichMapFunction<SensorReading, Integer>{
  21. //声明一个键控状态
  22. private ValueState<Integer> keyCountState;
  23. @Override
  24. public void open(Configuration parameters) throws Exception {
  25. keyCountState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("key-count", Integer.class));
  26. }
  27. @Override
  28. public Integer map(SensorReading value) throws Exception {
  29. //读取状态
  30. Integer count = keyCountState.value();
  31. count++;
  32. //对状态赋值
  33. keyCountState.update(count);
  34. return count;
  35. }
  36. }
  37. }

其它状态的示例代码:

  1. public class StateTest2_KeyedState {
  2. public static void main(String[] args) throws Exception {
  3. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4. env.setParallelism(1);
  5. // socket文本流
  6. DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
  7. // 转换成SensorReading类型
  8. // java8 中的lamda表达式
  9. DataStream<SensorReading> dataStream = inputStream.map(line -> {
  10. String[] fields = line.split(",");
  11. return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
  12. });
  13. //定义一个有状态的map操作,统计当前sensor数据个数
  14. SingleOutputStreamOperator<Integer> resultStream = dataStream.keyBy("id")
  15. .map(new MyKeyCountMapper());
  16. resultStream.print();
  17. env.execute();
  18. }
  19. //自定义实现一个RichMapFunction
  20. public static class MyKeyCountMapper extends RichMapFunction<SensorReading, Integer>{
  21. //声明一个键控状态
  22. private ValueState<Integer> keyCountState;
  23. //其它类型状态的声明
  24. private ListState<String> myListState;
  25. private MapState<String, Double> myMapState;
  26. private ReducingState<SensorReading> myReducingState;
  27. @Override
  28. public void open(Configuration parameters) throws Exception {
  29. keyCountState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("key-count", Integer.class));
  30. myListState = getRuntimeContext().getListState(new ListStateDescriptor<String>("my-list", String.class));
  31. myMapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Double>("my-map", String.class, Double.class));
  32. // myReducingState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<SensorReading>("my-reducing", , ));
  33. }
  34. @Override
  35. public Integer map(SensorReading value) throws Exception {
  36. //所有状态API都有clear()清空操作
  37. //其它状态API调用
  38. //list state
  39. for(String str : myListState.get()){
  40. System.out.println(str);
  41. }
  42. myListState.add("hello");
  43. //map state
  44. myMapState.get("key1");
  45. myMapState.put("key2", 35.001);
  46. myMapState.remove("key2");
  47. myMapState.clear();
  48. **加粗样式** //reducing state 聚合操作
  49. myReducingState.add(value);
  50. myReducingState.clear();
  51. //读取状态
  52. Integer count = keyCountState.value();
  53. count++;
  54. //对状态赋值
  55. keyCountState.update(count);
  56. return count;
  57. }
  58. }
  59. }

三、状态后端

  • 每传入一条数据,有状态的算子任务都会读取和更新状态
  • 由于有效的状态访问对于处理数据的低延迟至关重要,因此每个并行任务都会在本地维护其状态,以确保快速的状态访问
  • 状态的存储、访问及维护,由一个可插入的组件决定,这个组件就叫做状态后端(state backend)
  • 状态后端主要负责两件事:本地的状态管理以及检查点(checkpoint)状态写入远程存储

3.1 状态后端类型

  • MemoryStateBackend
    内存级的状态后端,会将键控状态作为内存中的对象进行管理,将它们存储在TaskManager的JVM堆上,而将checkpoint存储在JobManager的内存中
    特点: 快速、低延迟、但不稳定
  • FsStateBackend
    将checkpoint存到远程的持久化文件系统(FileSystem)上,而对于本地状态,跟MemoryStateBackend一样,也会存在TaskManager的JVM堆上
    特点: 同时拥有内存级的本地访问速度,和更好的容错保证
  • RocksDBStateBackend(适用于管理的状态多且对实时性要求不高的场景)
    将所有状态序列化后,存入本地的RocksDB中

3.2 状态后端的配置

方式一:
可以在flink安装包下找到 conf/flink-conf.yaml 文件,修改相应的状态后端配置

方式二:
可以在代码中进行配置

pom.xml

  1. <dependency>
  2. <groupId>org.apache.flink</groupId>
  3. <artifactId>flink-statebackend-rocksdb_2.12</artifactId>
  4. <version>1.10.1</version>
  5. </dependency>

代码

  1. public class StateTest3_FaultTolerance {
  2. public static void main(String[] args) throws Exception {
  3. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4. env.setParallelism(1);
  5. //1、设置状态后端
  6. env.setStateBackend(new MemoryStateBackend());
  7. env.setStateBackend(new FsStateBackend(""));
  8. env.setStateBackend(new RocksDBStateBackend(""));
  9. // socket文本流
  10. DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
  11. // 转换成SensorReading类型
  12. // java8 中的lamda表达式
  13. DataStream<SensorReading> dataStream = inputStream.map(line -> {
  14. String[] fields = line.split(",");
  15. return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
  16. });
  17. env.execute();
  18. }
  19. }

发表评论

表情:
评论列表 (有 0 条评论,268人围观)

还没有评论,来说两句吧...

相关阅读

    相关 Flink状态管理

    一、状态定义 由一个任务维护,并且用来计算某个结果的所有数据,都属于这个任务的状态 可以认为状态就是一个本地变量,可以被任务的业务逻辑访问 Flink会