Flink Keyed State 实例

短命女 2021-09-24 11:40 767阅读 0赞

**1 sourceStream必须要先keyBy然后才能使用Keyed State**

**2 需要继承RichxxxxFunction才行，在open之前声明，在open中初始化，在算子方法中使用和处理。不能继承xxxxxFunction，因为没有open方法，无法初始化，会报错。**

**3 open方法中只能初始化Keyed State，无法使用Keyed State（比如：获取值等操作），不然报错。因为open方法不属于Keyed上下文，算子方法才属于Keyed上下文**

并行度是1：2个不同key流被1个子任务处理

package StateTtlConfig;
    
    /**
     * @Author you guess
     * @Date 2020/6/22 22:42
     * @Version 1.0
     * @Desc
     */
    
    import org.apache.flink.api.common.functions.RichFlatMapFunction;
    import org.apache.flink.api.common.state.ValueState;
    import org.apache.flink.api.common.state.ValueStateDescriptor;
    import org.apache.flink.api.common.typeinfo.TypeHint;
    import org.apache.flink.api.common.typeinfo.TypeInformation;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    import org.apache.flink.util.Collector;
    
    import java.io.IOException;
    
    
    public class StateTest1 {
    
        public static void main(String[] args) throws Exception {
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStream<Tuple2<String, Long>> source1 = env.addSource(new SourceFunction<Tuple2<String, Long>>() {
                @Override
                public void run(SourceContext<Tuple2<String, Long>> ctx) throws Exception {
                    while (true) {
    
                        Thread.sleep(100);
                        ctx.collect(Tuple2.of("a", 1L));
    
                        Thread.sleep(100);
                        ctx.collect(Tuple2.of("b", 2L));
    
                        Thread.sleep(100);
                        ctx.collect(Tuple2.of("a", 3L));
    
                        Thread.sleep(100);
                        ctx.collect(Tuple2.of("b", 4L));
    
                        Thread.sleep(100);
                        ctx.collect(Tuple2.of("a", 5L));
    
                        Thread.sleep(100);
                        ctx.collect(Tuple2.of("b", 6L));
    
                        Thread.sleep(100);
                        ctx.collect(Tuple2.of("a", 7L));
    
                        Thread.sleep(100);
                        ctx.collect(Tuple2.of("b", 8L));
                        break;
                    }
                }
    
                @Override
                public void cancel() {
    
                }
            });//source1
    
            //KeyedState必须要用在keyBy后面！！不然报错！！
            //Keyed state can only be used on a 'keyed stream', i.e., after a 'keyBy()' operation.
            source1.keyBy(0)
                    .flatMap(new CountWindowAverage()).setParallelism(1)
                    .print();
    
            env.execute("Flink StateTTL Test1");
        }
    
    
        /**
         * KeyedState
         */
        public static class CountWindowAverage extends RichFlatMapFunction<Tuple2<String, Long>, Tuple2<String, Long>> {
    
            /**
             * The ValueState handle. The first field is the count, the second field a running sum.
             * 声明
             */
            ValueState<Tuple2<Long, Long>> averageKeyedState;
    
            @Override
            public void open(Configuration config) throws IOException {
                ValueStateDescriptor<Tuple2<Long, Long>> descriptor = new ValueStateDescriptor<Tuple2<Long, Long>>(
                        "average", // the state name
                        TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {
                        }), // type information
                        Tuple2.of(0L, 0L)); // default value of the state, if nothing was set
                //初始化
                averageKeyedState = getRuntimeContext().getState(descriptor);     
            
     //System.out.println("open...averageKeyedState:"+averageKeyedState.value());
                //NullPointerException: No key set. This method should not be called outside of a keyed context.
    
            }//open
    
            @Override
            public void flatMap(Tuple2<String, Long> input, Collector<Tuple2<String, Long>> out) throws Exception {
    
                // access the state value
                Tuple2<Long, Long> currentSum = averageKeyedState.value();
    
                // update the count
                currentSum.f0 += 1;
    
                // add the second field of the input value
                currentSum.f1 += input.f1;
    
                // update the state
                // 使用
                averageKeyedState.update(currentSum);
                System.out.println("sumKeyedState:" + averageKeyedState.value()+"，线程id#"+Thread.currentThread().getId());
    
                // if the count reaches 2, emit the average and clear the state
                if (currentSum.f0 >= 2) {
                    out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
                    averageKeyedState.clear();
                }
            }//flatMap
        }//CountWindowAverage
    }
    
    /*
    输出：
    sumKeyedState:(1,1)，线程id#72
    sumKeyedState:(1,2)，线程id#72
    sumKeyedState:(2,4)，线程id#72
    sumKeyedState:(2,6)，线程id#72
    8> (a,2)
    9> (b,3)
    sumKeyedState:(1,5)，线程id#72
    sumKeyedState:(1,6)，线程id#72
    sumKeyedState:(2,12)，线程id#72
    10> (a,6)
    sumKeyedState:(2,14)，线程id#72
    11> (b,7)
    * */

并行度是不设置（默认是12）：

//KeyedState必须要用在keyBy后面！！不然报错！！
            //Keyed state can only be used on a 'keyed stream', i.e., after a 'keyBy()' operation.
            source1.keyBy(0)
                    .flatMap(new CountWindowAverage())//.setParallelism(1)
                    .print();
    
    /*
    输出：
    sumKeyedState:(1,1)，线程id#81
    sumKeyedState:(1,2)，线程id#74
    sumKeyedState:(2,4)，线程id#81
    8> (a,2)
    sumKeyedState:(2,6)，线程id#74
    3> (b,3)
    sumKeyedState:(1,5)，线程id#81
    sumKeyedState:(1,6)，线程id#74
    sumKeyedState:(2,12)，线程id#81
    8> (a,6)
    sumKeyedState:(2,14)，线程id#74
    3> (b,7)
    */

flink1.9.2,java1.8