直接上代码如下：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class KeyedStateDeepReview {
    public static void main(String[] args) throws Exception {
        //1.创建一个 flink steam 程序的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 重启相关配置
        // 只有开启了checkpointing,才会有重启策略
        env.enableCheckpointing(5000);    // 5秒为一个周期
//        // 设置最多重启3次，每次间隔两秒
        // 默认的重启策略是  延迟无限重启CheckpointConfig
        env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000));
        //为了实现Exactly_once,必须呀记录偏移量
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //3. 创建DataStream
        DataStreamSource<String> lines = env.socketTextStream( "192.168.***.***", 8888);
        //Transformation（s） 对数据进行处理操作
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String word) {
                //将每个单词与 1 组合，形成一个元组
                return Tuple2.of(word, 1);
        //4. Transformation 进行分组聚合(keyBy：将key相同的分到一个组中)
        KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = wordAndOne.keyBy(0);
        // 使用 KeyedState 通过中间状态求和
        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = keyedStream.map(new KeyedStateSum());
        //5.调用Sink （Sink必须调用）
        summed.print();
        //6. 启动
        env.execute("KeyedStateDeepReview");
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.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
public class KeyedStateSum extends RichMapFunction<Tuple2<String,Integer>, Tuple2<String,Integer>> {
    //状态数据不参与序列化，添加 transient 修饰
    private transient ValueState<Tuple2<String,Integer>> valueState;
    // open方法在完成构造方法后执行一次
    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Tuple2<String,Integer>> stateDescriptor =
                new ValueStateDescriptor<>("wc-keyed-state", Types.TUPLE(Types.STRING, Types.INT));
        // 每一个组key都有自己的状态，无需特别指定是哪个word组的状态
        valueState = getRuntimeContext().getState(stateDescriptor);
    @Override
    public Tuple2<String, Integer> map(Tuple2<String, Integer> value) throws Exception {
        //输入的单词
        String word = value.f0;
        //输入的次数
        Integer count = value.f1;
        //根据State获取中间数据
        Tuple2<String,Integer> historyKV = valueState.value();
        //根据State中间数据，进行累加
        if (historyKV != null) {
            historyKV.f1 += count;
            //累加后,更新State数据
            valueState.update(historyKV);
        } else {
            // 更新历史数据
            valueState.update(value);
//                return historyKV;  //这样写会报空指针异常
//                return Tuple2.of(word,valueState.value().f1);
        return valueState.value();
测试结果：成功实现累加功能
 
                    直接上代码如下：import org.apache.flink.api.common.functions.MapFunction;import org.apache.flink.api.common.restartstrategy.RestartStrategies;import org.apache.flink.api.java.tuple.Tuple;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.
				
       /* &lt;pre&gt;{@code
     * DataStream&lt;MyType&gt; stream = ...;
     * KeyedStream&lt;MyType&gt; keyedStream = stream.keyBy("id");
     * keyedStream.map(new RichMapFunction&lt;My...
				
键控状态是根据输入数据流中定义的键（key）来维护和访问的。Flink 为每个键值维护一个状态实例，并将具有相同键的所有数据，都分区到同一个算子任务中，这个任务会维护和处理这个 key 对应的状态。当任务处理一条数据时，它会自动将状态的访问范围限定为当前数据的 key。因此，具有相同 key 的所有数据都会访问相同的状态。Keyed State 很类似于一个分布式的 key-value map 数据结构，只能用于 KeyedStream（keyBy 算子处理之后）。
Flink 的 Keyed State
package cn._51doit.flink.day06;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy..
				
文章目录基本概念和特点支持的结构类型代码实现
在实际应用中，我们一般都需要将数据按照某个 key 进行分区，然后再进行计算处理；所以最为常见的状态类型就是 Keyed State。
另外，我们还可以通过富函数类（Rich Function）对转换算子进行扩展、实现自定义功能，比如 RichMapFunction、RichFilterFunction。在富函数中，我们可以调用.getRuntimeContext()获取当前的运行时上下文（RuntimeContext），进而获取到访问状态的句柄；这种富函数中自
public class KeyedStateDemo {
    public static void main(String[] args) throws Exception {
        //1.创建一个 flink steam 程序的执行环境
本文主要研究一下flink的Managed Keyed State
State
flink-core-1.7.0-sources.jar!/org/apache/flink/api/common/state/State.java
 * Interface that different types of partitioned state must implement.
 * &l...
public class WordCountKeyedState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 初始化测试单词数据流
        DataStreamSource<S...
				
KeyedStateStore有ValueState，ListState，ReducingState，AggregatingState，MapState5中状态。下面实现这5种状态示例。
1.KeyedStateStore接口的所有get方法
2.ValueState
  DataStream<MyType> stream = ...;
	 KeyedStream<MyType...
				
1. 累加器的简单介绍
累加器是从用户函数和操作中，分布式地统计或者聚合信息。每个并行实例创建并更新自己的Accumulator对象, 然后合并收集器的不同并行实例。在作业结束时由系统合并。
累加器的结果可以从作业执行的结果中获得，也可以从Web运行时监视器中获得。
累加器是受Hadoop/MapReduce计数器的启发。但是要注意添加到累加器的类型可能与返回的类型不同。比如：我们添加单个对象,但...
            007_centos使用paddlepaddle报错ImportError: /lib64/libstdc++.so.6: version `GLIBCXX_3.4.20‘ not found
            linux部署Python项目，并解决依赖自定义模块报错问题
            RPM命令的--nodeps 和--force参数解释