谢了
在 2023-05-06 10:36:02，"Hangxiang Yu" <master...@gmail.com> 写道：
>Hi, 在 initializeState 里初始化 State 是 OK 的，但是尽量不要在 initializeState 和
>snapshotState 里访问 KeyedState，最好是在实际的 Function 比如这里的 FlatMap 里；
>原因是 KeyedState 的访问是和 Current Key 绑定的，实际的 Function 在 Process 之前会隐式的 set
>Current key ，因此是会保证每次 KeyedState 的操作是对确定的 KV 进行的；
>而 initializeState 和 snapshotState 里是没有框架隐性 set 的，相当于会对某一个不确定的 key 去update
>value了，如果一定要在这里做就需要拿到 KeyContext 自己去 set，不过不建议这么做；
>
>On Fri, May 5, 2023 at 10:58 PM sjf0115 <si_ji_f...@163.com> wrote:
>
>> CheckpointedFunction 接口的 initializeState 方法提供了访问
>> FunctionInitializationContext 的能力，而 FunctionInitializationContext 不仅提供了访问
>> OperatorStateStore 的能力，也提供了 KeyedStateStore 的能力。一般常见的是通过
>> CheckpointedFunction 来实现操作 OperatorState，但是也可以通过如下代码来获取 KeyedState：
>> ```java
>> context.getKeyedStateStore().getState(stateDescriptor);
>> ```
>> 想问一下通过 CheckpointedFunction 来实现操作 KeyedState，如下代码所示。建议这样实现吗？会有什么问题吗？
>> ```java
>> public static class TemperatureAlertFlatMapFunction extends
>> RichFlatMapFunction<Tuple2<String, Double>, Tuple3<String, Double, Double>>
>> implements CheckpointedFunction {
>> // 温度差报警阈值
>> private double threshold;
>> // 上一次温度
>> private ValueState<Double> lastTemperatureState;
>> private Double lastTemperature;
>> public TemperatureAlertFlatMapFunction(double threshold) {
>> this.threshold = threshold;
>> }
>>
>>
>> @Override
>> public void flatMap(Tuple2<String, Double> sensor,
>> Collector<Tuple3<String, Double, Double>> out) throws Exception {
>> String sensorId = sensor.f0;
>> // 当前温度
>> double temperature = sensor.f1;
>> // 保存当前温度
>> lastTemperature = temperature;
>> // 是否是第一次上报的温度
>> if (Objects.equals(lastTemperature, null)) {
>> return;
>> }
>> double diff = Math.abs(temperature - lastTemperature);
>> if (diff > threshold) {
>> // 温度变化超过阈值则输出
>> out.collect(Tuple3.of(sensorId, temperature, diff));
>> }
>> }
>>
>>
>> @Override
>> public void snapshotState(FunctionSnapshotContext context) throws
>> Exception {
>> // 获取最新的温度之后更新保存上一次温度的状态
>> if (!Objects.equals(lastTemperature, null)) {
>> lastTemperatureState.update(lastTemperature);
>> }
>> }
>>
>>
>> @Override
>> public void initializeState(FunctionInitializationContext context)
>> throws Exception {
>> ValueStateDescriptor<Double> stateDescriptor = new
>> ValueStateDescriptor<>("lastTemperature", Double.class);
>> lastTemperatureState =
>> context.getKeyedStateStore().getState(stateDescriptor);
>> if (context.isRestored()) {
>> lastTemperature = lastTemperatureState.value();
>> }
>> }
>> }
>> ```
>>
>>
>
>--
>Best,
>Hangxiang.
