聊聊Flink框架中的状态管理机制

百思不得小赵 2023-03-28 原文

状态概述

在目前所有流式计算的场景中，将数据流的状态分为有状态和无状态两种类型。无状态指的就是无状态的计算观察每个独立的事件，并且只根据最后一个事件输出结果。举个栗子：一个流处理程序，从传感器接收温度数据然后在温度为90摄氏度发出报警信息。有状态的计算则会根据多个事件输出结果。举个栗子：计算过去一小时的平均温度，就是有状态的计算、若在一分钟内收到两个相差 20 度以上的温度读数，则发出警告等等。

对照上图可以看出：

无状态流处理分别接收每条数据记录，然后根据最新输入的数据生成输出数据。（每次只转换一条输入记录，并且仅根据最新的输入记录输出结果）
有状态流处理会维护状态，并基于最新输入的记录和当前的状态值生成输出记录。（维护所有已处理记录的状态值，并根据每条新输入的记录更新状态，因此输出记录反映的是综合考虑多个事件之后的结果。）

Flink中的状态

Flink中的状态有一个任务进行专门维护，并且用来计算某个结果的所有数据，都属于这个任务的状态。大多数的情况下我们可以将Flink中状态理解为一个本地变量，存储在内存中。状态自始至终是与特定的算子相关联的，在flink中需要进行状态的注册。

（此图来源于网络）

Flink框架中有两种类型的状态：算子状态、键控状态。接下来我们具体的聊聊这两种状态。

算子状态

算子状态的作用范围限定为算子任务。由同一并行任务所处理的所有数据都可以访问到相同的状态。也就是说，同一个任务中是共享的。
注意：算子状态不能由相同或不同算子的另一个子任务访问

（此图来源于网络）

Flink 为算子状态提供三种基本数据结构：

列表状态

将状态表示为一组数据的列表。

联合列表状态

也将状态表示为数据的列表。它与常规列表状态的区别在于，在发生故障时，或者从保存点（savepoint）启动应用程序时如何恢复。

广播状态

如果一个算子有多项任务，而它的每项任务状态又都相同，那么这种特殊情况最适合应用广播状态

代码如下：

public class StateTest1_OperatorState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设施并行度为1
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 7777);
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // 定义一个有状态的Map操作  统计当前分区数据个数
        SingleOutputStreamOperator<Integer> resultStream = dataStream.map(new MyCountMap());

        resultStream.print();
        env.execute();
    }

    //自定义mapFunction 注册状态实现ListCheckpointed
    private static class MyCountMap implements MapFunction<SensorReading, Integer>, ListCheckpointed<Integer>{

        // 定义一个本地变量作为算子状态
        private Integer count = 0;

        @Override
        public Integer map(SensorReading sensorReading) throws Exception {
            count++;
            return count;
        }

        @Override
        public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
            return Collections.singletonList(count);
        }

        @Override
        public void restoreState(List<Integer> state) throws Exception {
            for (Integer number : state) {
                count+=number;
            }
        }
    }
}

键控状态

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

（此图来源于网络）

Flink 为键控状态提供三种基本数据结构：

值状态

将状态表示为单个的值。

列表状态

将状态表示为一组数据的列表

映射状态

将状态表示为一组 Key-Value 对

聚合状态（Reducing state & Aggregating State）

将状态表示为一个用于聚合操作的列表

代码如下：

public class StateTest2_KeyedState {
    public static void main(String[] args)  throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设施并行度为1
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 7777);
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // 定义一个有状态的Map操作  统计当前senser数据个数
        SingleOutputStreamOperator<Integer> resultStream = dataStream.keyBy("id")
                .map(new MyKeyCouneMap());

        resultStream.print();
        env.execute();
    }

    // 自定义RichMapFunction
    private static class MyKeyCouneMap extends RichMapFunction<SensorReading,Integer> {
         // 声明键控状态
         private ValueState<Integer> KeyCouneState ;

        //private ListState listState;

        //private MapState mapState;

       // private ReducingState reduceState;

        @Override
        public void open(Configuration parameters) throws Exception {
            KeyCouneState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>(
                    "key-count",Integer.class
            ));

           // listState =getRuntimeContext().getListState(new ListStateDescriptor(
                 //   "list-count",String.class
            ));

            //mapState = getRuntimeContext().getMapState(new MapStateDescriptor(
                 //   "map-count",String.class,Double.class
            ));

            //reduceState =getRuntimeContext().getReducingState(new ReducingStateDescriptor(
            //        "reducing-count",SensorReading.class
            //));class
        }

        @Override
        public Integer map(SensorReading sensorReading) throws Exception {
            // 读取状态
            Integer count = KeyCouneState.value();
            if (count==null){
                count = 0;
            }else {
                count++;
            }
            // 更新状态，对状态赋值
            KeyCouneState.update(count);
            return count;

            // listState
            //Iterable iterable = listState.get();
            //for (String s : listState.get()) {
            //    System.out.println(s);
            //}
            //listState.add("hello");

            // mapState
            //mapState.get("1");
            //mapState.put("2",12.3);

        }
    }
}

状态后端

状态的存储、访问以及维护，由一个可插入的组件决定，这个组件就叫做状态后端。状态后端主要负责两件事：本地的状态管理，以及将检查点（checkpoint）状态写入远程存储。状态后端总共有三种类型：

MemoryStateBackend

内存级的状态后端，会将键控状态作为内存中的对象进行管理，将它们存储在TaskManager 的 JVM 堆上，而将 checkpoint 存储在JobManager 的内存中。特点：快速、低延迟，但不稳定。

FsStateBackend

将 checkpoint 存到远程的持久化文件系统（FileSystem）上，而对于本地状态，跟 MemoryStateBackend 一样，也会存在 TaskManager 的 JVM 堆上同时拥有内存级的本地访问速度，和更好的容错保证。

RocksDBStateBackend

将所有状态序列化后，存入本地的 RocksDB 中存储

代码中配置状态后端：

public class StateTest4_FaultTolerance {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 状态后端配置
        env.setStateBackend(new MemoryStateBackend());
        env.setStateBackend(new FsStateBackend(""));
        
        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 7777);
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });
        env.execute();
    }
}

flink配置文件中进行配置：

一个案例：

检查工业物联网传感器温度跳变，如果连续两个温度差值超过10度，就发出报警。

代码如下

public class StateTest3_ApplicationCase {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设施并行度为1
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 7777);
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // 定义一个flutmap操作，检测温度跳变输出报警
        SingleOutputStreamOperator<Tuple3<String, Double, Double>> resultStream = dataStream.keyBy("id")
                .flatMap(new TempChangeWarring(10.0));

        resultStream.print();
        env.execute();
    }

    private static class TempChangeWarring extends RichFlatMapFunction<SensorReading, Tuple3<String, Double, Double>> {

        // 设置温度跳变的阈值
        private Double threshold;

        public TempChangeWarring(Double threshold) {
            this.threshold = threshold;
        }

        // 定义状态，保存上一次温度值
        private ValueState<Double> lastTempState;

        @Override
        public void open(Configuration parameters) throws Exception {
            lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>(
                    "lastTemp", Double.class
            ));
        }

        @Override
        public void flatMap(SensorReading value, Collector<Tuple3<String, Double, Double>> out) throws Exception {
            // 获取状态
            Double lastTemp = lastTempState.value();
            // 如果不为null 就计算两次温度差
            if (lastTemp != null) {
                Double diff = Math.abs(value.getTemperature() - lastTemp);
                if (diff>=threshold){
                    out.collect(new Tuple3<>(value.getId(),lastTemp,value.getTemperature()));
                }
            }

            // 更新状态
            lastTempState.update(value.getTemperature());
        }

        @Override
        public void close() throws Exception {
            lastTempState.clear();
        }
    }
}

Flink 聊聊 span style color flink bigdata java 经验分享数据OpenStack 云计算

有关聊聊Flink框架中的状态管理机制的更多相关文章

ruby - 如何从 ruby 中的字符串运行任意对象方法？ - 2
总的来说，我对ruby还比较陌生，我正在为我正在创建的对象编写一些rspec测试用例。许多测试用例都非常基础，我只是想确保正确填充和返回值。我想知道是否有办法使用循环结构来执行此操作。不必为我要测试的每个方法都设置一个assertEquals。例如:describeitem,"TestingtheItem"doit"willhaveanullvaluetostart"doitem=Item.new#HereIcoulddotheitem.name.shouldbe_nil#thenIcoulddoitem.category.shouldbe_nilendend但我想要一些方法来使用
ruby - 其他文件中的 Rake 任务 - 2
我试图在一个项目中使用rake，如果我把所有东西都放到Rakefile中，它会很大并且很难读取/找到东西，所以我试着将每个命名空间放在lib/rake中它自己的文件中，我添加了这个到我的rake文件的顶部:Dir['#{File.dirname(__FILE__)}/lib/rake/*.rake'].map{|f|requiref}它加载文件没问题，但没有任务。我现在只有一个.rake文件作为测试，名为“servers.rake”，它看起来像这样:namespace:serverdotask:testdoputs"test"endend所以当我运行rakeserver:testid时
ruby-on-rails - Ruby net/ldap 模块中的内存泄漏 - 2
作为我的Rails应用程序的一部分，我编写了一个小导入程序，它从我们的LDAP系统中吸取数据并将其塞入一个用户表中。不幸的是，与LDAP相关的代码在遍历我们的32K用户时泄漏了大量内存，我一直无法弄清楚如何解决这个问题。这个问题似乎在某种程度上与LDAP库有关，因为当我删除对LDAP内容的调用时，内存使用情况会很好地稳定下来。此外，不断增加的对象是Net::BER::BerIdentifiedString和Net::BER::BerIdentifiedArray，它们都是LDAP库的一部分。当我运行导入时，内存使用量最终达到超过1GB的峰值。如果问题存在，我需要找到一些方法来更正我的代
ruby - i18n Assets 管理/翻译 UI - 2
我正在使用i18n从头开始构建一个多语言网络应用程序，虽然我自己可以处理一大堆yml文件，但我说的语言(非常)有限，最终我想寻求外部帮助帮助。我想知道这里是否有人在使用UI插件/gem(与django上的django-rosetta不同)来处理多个翻译器，其中一些翻译器不愿意或无法处理存储库中的100多个文件，处理语言数据。谢谢&问候，安德拉斯(如果您已经在rubyonrails-talk上遇到了这个问题，我们深表歉意) 最佳答案有一个rails3branchofthetolkgem在github上。您可以通过在Gemfi
ruby - 在 Ruby 程序执行时阻止 Windows 7 PC 进入休眠状态 - 2
我需要在客户计算机上运行Ruby应用程序。通常需要几天才能完成(复制大备份文件)。问题是如果启用sleep，它会中断应用程序。否则，计算机将持续运行数周，直到我下次访问为止。有什么方法可以防止执行期间休眠并让Windows在执行后休眠吗？欢迎任何疯狂的想法;-) 最佳答案 Here建议使用SetThreadExecutionStateWinAPI函数，使应用程序能够通知系统它正在使用中，从而防止系统在应用程序运行时进入休眠状态或关闭显示。像这样的东西:require'Win32API'ES_AWAYMODE_REQUIRED=0x0
ruby-on-rails - Rails 3 中的多个路由文件 - 2
Rails2.3可以选择随时使用RouteSet#add_configuration_file添加更多路由。是否可以在Rails3项目中做同样的事情？最佳答案在config/application.rb中:config.paths.config.routes在Rails3.2(也可能是Rails3.1)中，使用:config.paths["config/routes"] 关于ruby-on-rails-Rails3中的多个路由文件，我们在StackOverflow上找到一个类似的问题
ruby-on-rails - Rails - 一个 View 中的多个模型 - 2
我需要从一个View访问多个模型。以前，我的links_controller仅用于提供以不同方式排序的链接资源。现在我想包括一个部分(我假设)显示按分数排序的顶级用户(@users=User.all.sort_by(&:score))我知道我可以将此代码插入每个链接操作并从View访问它，但这似乎不是“ruby方式”，我将需要在不久的将来访问更多模型。这可能会变得很脏，是否有针对这种情况的任何技术？注意事项:我认为我的应用程序正朝着单一格式和动态页面内容的方向发展，本质上是一个典型的网络应用程序。我知道before_filter但考虑到我希望应用程序进入的方向，这似乎很麻烦。最终从任何
ruby-on-rails - Rails 3.2.1 中 ActionMailer 中的未定义方法 'default_content_type=' - 2
我在我的项目中添加了一个系统来重置用户密码并通过电子邮件将密码发送给他，以防他忘记密码。昨天它运行良好(当我实现它时)。当我今天尝试启动服务器时，出现以下错误。=>BootingWEBrick=>Rails3.2.1applicationstartingindevelopmentonhttp://0.0.0.0:3000=>Callwith-dtodetach=>Ctrl-CtoshutdownserverExiting/Users/vinayshenoy/.rvm/gems/ruby-1.9.3-p0/gems/actionmailer-3.2.1/lib/action_mailer
ruby-on-rails - Rails 应用程序中的 Rails : How are you using application_controller. rb 是新手吗？ - 2
刚入门rails，开始慢慢理解。有人可以解释或给我一些关于在application_controller中编码的好处或时间和原因的想法吗？有哪些用例。您如何为Rails应用程序使用应用程序Controller？我不想在那里放太多代码，因为据我了解，每个请求都会调用此Controller。这是真的？最佳答案 ApplicationController实际上是您应用程序中的每个其他Controller都将从中继承的类(尽管这不是强制性的)。我同意不要用太多代码弄乱它并保持干净整洁的态度，尽管在某些情况下ApplicationContr
ruby-on-rails - form_for 中不在模型中的自定义字段 - 2
我想向我的Controller传递一个参数，它是一个简单的复选框，但我不知道如何在模型的form_for中引入它，这是我的观点:{:id=>'go_finance'}do|f|%>Transferirde:para:Entrada:"input",:placeholder=>"Quantofoiganho?"%>Saída:"output",:placeholder=>"Quantofoigasto?"%>Nota:我想做一个额外的复选框，但我该怎么做，模型中没有一个对象，而是一个要检查的对象，以便在Controller中创建一个ifelse，如果没有检查，请帮助我，非常感谢,谢谢

聊聊Flink框架中的状态管理机制

目录

状态概述

算子状态

键控状态

状态后端

有关聊聊Flink框架中的状态管理机制的更多相关文章

随机推荐