Flink从入门到实战八[State]-4-Flink Keyed State ValueState新版API使用方法

我们来看一下使用ValueStateDescriptor和ValueState，基于状态如何累计计算一个用户的交易数据，demo是以较新的1.14.4版本进行演示。

使用ValueState重点关注三点：
1、ValueState的方法： update(T)、T value()、clear()
2、状态描述符的创建： ValueStateDescriptor
3、状态注册：ValueState getState(ValueStateDescriptor)

代码：

mport java.util.Objects;

/**
 * 基于Flink状态求和：按照用户id分组求每个人的消费总笔数、消费总金额
 * flink version 1.14.4
 * ValueStateDescriptor使用新版本定义使用方法演示
 */
public class State_2_ValueState {

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

        DataStream<Transaction> dataStream = env
                .addSource(new TransactionSource()) //TransactionSource可以查看前面章节，有源码分析讲解
                .name("transactions");
        dataStream.print();

        dataStream.keyBy(Transaction::getAccountId)
                .flatMap(new State_2_FlatMap())
                .print();

        env.execute();
    }

    public static class State_2_FlatMap extends RichFlatMapFunction<Transaction, Tuple3<Long, Integer, Double>> {

        private ValueState<Tuple2<Integer, Double>> sum;

        @Override
        public void open(Configuration parameters) throws Exception {
			//新版ValueStateDescriptor取消了初始化状态的构造器
            ValueStateDescriptor<Tuple2<Integer, Double>> descriptor =
                    new ValueStateDescriptor<>(
                            "sum", Types.TUPLE(TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {})));
            sum = getRuntimeContext().getState(descriptor);
        }

        @Override
        public void flatMap(Transaction transaction, Collector<Tuple3<Long, Integer, Double>> collector) throws Exception {
            Tuple2<Integer, Double> value = sum.value();
            //在业务代码中判断state是否初始化
            if(Objects.isNull(value)) {
                value = new Tuple2<>(0,0d);
            }
            value.f0 = value.f0+1;
            value.f1 = value.f1+transaction.getAmount();
            sum.update(value);
            collector.collect(new Tuple3<>(transaction.getAccountId(), value.f0, value.f1));
        }
    }
}
/* 输出结果
3> Transaction{accountId=1, timestamp=1546272000000, amount=188.23}
6> (1,1,188.23)
4> Transaction{accountId=2, timestamp=1546272360000, amount=374.79}
8> (2,1,374.79)
5> Transaction{accountId=3, timestamp=1546272720000, amount=112.15}
8> (3,1,112.15)
6> Transaction{accountId=4, timestamp=1546273080000, amount=478.75}
1> (4,1,478.75)
7> Transaction{accountId=5, timestamp=1546273440000, amount=208.85}
8> (5,1,208.85)
8> Transaction{accountId=1, timestamp=1546273800000, amount=379.64}
6> (1,2,567.87)
1> Transaction{accountId=2, timestamp=1546274160000, amount=351.44}
8> (2,2,726.23)
2> Transaction{accountId=3, timestamp=1546274520000, amount=320.75}
8> (3,2,432.9)
3> Transaction{accountId=4, timestamp=1546274880000, amount=259.42}
1> (4,2,738.1700000000001)
4> Transaction{accountId=5, timestamp=1546275240000, amount=273.44}
8> (5,2,482.28999999999996)
5> Transaction{accountId=1, timestamp=1546275600000, amount=267.25}
6> (1,3,835.12)
6> Transaction{accountId=2, timestamp=1546275960000, amount=397.15}
8> (2,3,1123.38)
7> Transaction{accountId=3, timestamp=1546276320000, amount=0.219}
8> (3,3,433.11899999999997)
8> Transaction{accountId=4, timestamp=1546276680000, amount=231.94}
1> (4,3,970.1100000000001)
1> Transaction{accountId=5, timestamp=1546277040000, amount=384.73}
8> (5,3,867.02)
2> Transaction{accountId=1, timestamp=1546277400000, amount=419.62}
6> (1,4,1254.74)
3> Transaction{accountId=2, timestamp=1546277760000, amount=412.91}
8> (2,4,1536.2900000000002)
 */

Flink State 使用流程总结分为四步：
1、注册状态
2、读取状态
3、更新状态
4、清除状态（可选的步骤）