ToB企服应用市场:ToB评测及商务社交产业平台

标题: 大数据-124 - Flink State 01篇 状态原理和原理分析：状态范例 实行分析 [打印本页]
作者: 卖不甜枣    时间: 2024-9-10 23:23
标题: 大数据-124 - Flink State 01篇 状态原理和原理分析：状态范例 实行分析
点一下关注吧！！！非常感谢！！连续更新！！！

目前已经更新到了：


章节内容

上节我们完成了如下的内容：

状态范例

Flink根据是否需要保存中间结果，把计算分为有状态计算和无状态计算。

根据数据布局差别，Flink定义了多种State，应用于差别的场景。

State按照是否有Key划分为：

案例1 使用State求均匀值

实现思路


编写代码

  1. package icu.wzk;
  3. import org.apache.flink.api.common.functions.RichFlatMapFunction;
  4. import org.apache.flink.api.common.state.ValueState;
  5. import org.apache.flink.api.common.state.ValueStateDescriptor;
  6. import org.apache.flink.api.common.typeinfo.TypeHint;
  7. import org.apache.flink.api.common.typeinfo.TypeInformation;
  8. import org.apache.flink.api.java.functions.KeySelector;
  9. import org.apache.flink.api.java.tuple.Tuple2;
  10. import org.apache.flink.configuration.Configuration;
  11. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  12. import org.apache.flink.streaming.api.datastream.KeyedStream;
  13. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  14. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  15. import org.apache.flink.util.Collector;
  18. public class FlinkStateTest01 {
  20.     public static void main(String[] args) throws Exception {
  21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  22.         env.setParallelism(1);
  23.         DataStreamSource<Tuple2<Long, Long>> data = env
  24.                 .fromElements(
  25.                         Tuple2.of(1L, 3L),
  26.                         Tuple2.of(1L, 5L),
  27.                         Tuple2.of(1L, 7L),
  28.                         Tuple2.of(1L, 4L),
  29.                         Tuple2.of(1L, 2L)
  30.                 );
  31.         KeyedStream<Tuple2<Long, Long>, Long> keyed = data
  32.                 .keyBy(new KeySelector<Tuple2<Long, Long>, Long>() {
  33.                     @Override
  34.                     public Long getKey(Tuple2<Long, Long> value) throws Exception {
  35.                         return value.f0;
  36.                     }
  37.                 });
  38.         SingleOutputStreamOperator<Tuple2<Long, Long>> flatMapped = keyed
  39.                 .flatMap(new RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>() {
  40.                     private transient ValueState<Tuple2<Long, Long>> sum;
  42.                     @Override
  43.                     public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
  44.                         Tuple2<Long, Long> currentSum = sum.value();
  45.                         if (currentSum == null) {
  46.                             currentSum = Tuple2.of(0L, 0L);
  47.                         }
  48.                         // 更新
  49.                         currentSum.f0 += 1L;
  50.                         currentSum.f1 += value.f1;
  51.                         System.out.println("currentValue: " + currentSum);
  52.                         // 更新状态值
  53.                         sum.update(currentSum);
  54.                         // 如果 count >= 5 清空状态值 重新计算
  55.                         if (currentSum.f0 >= 5) {
  56.                             out.collect(new Tuple2<>(value.f0, currentSum.f1 / currentSum.f0));
  57.                             sum.clear();
  58.                         }
  59.                     }
  61.                     @Override
  62.                     public void open(Configuration parameters) throws Exception {
  63.                         ValueStateDescriptor<Tuple2<Long, Long>> descriptor = new ValueStateDescriptor<>(
  64.                                 "average",
  65.                                 TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {})
  66.                         );
  67.                         sum = getRuntimeContext().getState(descriptor);
  68.                     }
  69.                 });
  70.         flatMapped.print();
  71.         env.execute("Flink State Test");
  72.     }
  73. }
复制代码
运行结果


实行分析



Keyed State

表示和Key相干的一种State， 只能用于KeyedStream范例数据集对应的Function和Operator之上，KeyedState是OperatorState的特例，区别在于KeyedState事先按照Key对数据集进行了区分，每个KeyState仅对应一个Operator和Key的组合。
KeyedState可以通过KeyGroups进行管理，主要用于当算子并行度发生变革时，自动重新分布KeyedState数据。在系统运行过程中，一个Keyed算子实例大概运行一个大概多个KeyGroups的Keys。
Operator State

与 Keyed State 差别的是，Operator State 只和并行的算子实例绑定，和数据元素中的Key无关，每个算子实例中持有全部数据元素中的一部分状态数据。Operator State 支持算子实例并行度发生变革时自动重新分配状态数据。
同时在Flink中KeyedState和OperatorState均具有两种形式，此中一种为托管状态（Managed State）形式，由FlinkRuntime中控制和管理状态数据，并将状态数据转换为内存HashTables或RocksDB的对象存储，然后将这些状态数据通过内部的接口长期话到CheckPoints中，使命非常时可以通过这些状态数据恢复使命。另外一种是原生状态（Row State）形式，由算子本身管理数据布局，当触发CheckPoint中，当从CheckPoint恢复使命时，算子本身再返序列化出状态的数据布局。
DataStreamAPI支持使用ManagedState和RawState两种状态形式，在Flink中推荐用户使用ManagedState管理状态数据，主要缘故原由是ManagedState可以或许更好地支持状态数据的重平衡以及更加完善的内存管理。
状态描述


State既然是暴露给用户的，那么就需要有一些属性需要指定：

在对应的StateBackend中，会去调用对应的create方法获取到stateDescriptor中的值。
Flink通过StateDescriptor来定义一个状态，这是一个抽象类，内部定义了状态名称、范例、序列化器等底子信息，与上面的状态对应，从StateDescriptor派生ValueStateDescriptor、ListStateDescriptor等等


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4