那我理解了,描述的太清晰了,我确实之前没理解state的生命周期和使用方法,十分感谢
在 2022-08-29 18:42:51,"Zhiwen Sun" <pens...@gmail.com> 写道: >你应该没有正确理解 state 的使用 > >我们一般在程序里面是用的是 KeyedState , 也就是和 key 伴随的。 > >基于上面,所以 open() 里面只能对 state 进行初始化, 但是没有办法设置 state 的 value,因为这时候没有 key ; >另外一方面,也不会在 map() 的时候去 new state (可以认为 state 是一个大的 Map,你 map 的时候只是操作其中的一个 >key)。 > >回到你的需求,你应该在 open() 的时候保存相关信息到类变量里面,当 map() 的时候再去 update state。 > > > > > > >Zhiwen Sun > > > >On Fri, Aug 26, 2022 at 1:55 PM 曲洋 <quyanghao...@126.com> wrote: > >> >> 对的,是后者,statAccumulator.value()是null,就是map方法中取值就成null了,但是open中命名初始化了,这个是因为map太快了吗,state没初始化完就开始拿了吗. >> 嗯嗯,我现在改成先进行判断 >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> 在 2022-08-26 11:22:43,"Hangxiang Yu" <master...@gmail.com> 写道: >> >open确实是初始化的时候就会调用的; >> >> >第一次调用是null是说statAccumulator是null还是statAccumulator.value()是null,后者的话是正常可能会出现的; >> >这里的写法看起来有点问题,一般用value方法取出来可以先判断下,然后对value state的更新用update方法; >> > >> >On Fri, Aug 26, 2022 at 10:25 AM 曲洋 <quyanghao...@126.com> wrote: >> > >> >> 各位好, >> >> >> >> >> 我想请教一个问题,我的Flink应用中会在state里边存储一些指标,比如一年的总数,然后当任务各种原因断掉的时候,我希望可以通过入参的方式直接调节这个state,但是遇到了一个问题,如下: >> >> 我重写了RichMapFunction,yearTotal >> >> >> 这个指标是通过命令行传进来的,然后我希望初始化到state里边,但是我发现,open方法第一次调用的时候state都是null,然后这个参数就进不来 >> >> 所以我想问下这个场景怎么办,还有open方法的生命周期是怎么样的,我本以为是map第一次打开的时候就会调用,结果好像不是 >> >> public static class AccumulateAmounts extends RichMapFunction<v2bean, >> >> BlueAccumulaterInitState> { >> >> private transient ValueState<BlueAccumulaterInitState> >> >> statAccumulator; >> >> >> >> >> >> @Override >> >> public BlueAccumulaterInitState map(v2bean currentAccumulator) >> >> throws Exception { >> >> >> >> >> >> BlueAccumulaterInitState stat = (statAccumulator.value() != >> >> null) ? statAccumulator.value() : new BlueAccumulaterInitState(); >> >> Long yearIncrement = year.equals(stat.getYear()) ? >> >> stat.getYearMetric() + 1L : 1L; >> >> stat.setYearMetric(yearIncrement); >> >> >> >> >> >> statAccumulator.update(stat); >> >> return stat; >> >> } >> >> >> >> >> >> @Override >> >> public void open(Configuration config) { >> >> ValueStateDescriptor<BlueAccumulaterInitState> descriptor = >> >> new ValueStateDescriptor<>( >> >> "total", >> >> TypeInformation.of(new >> >> TypeHint<BlueAccumulaterInitState>() { >> >> })); >> >> statAccumulator = getRuntimeContext().getState(descriptor); >> >> ExecutionConfig.GlobalJobParameters globalParams = >> >> getRuntimeContext().getExecutionConfig().getGlobalJobParameters(); >> >> Configuration globConf = (Configuration) globalParams; >> >> long yearTotal = >> >> globConf.getLong(ConfigOptions.key("year").longType().noDefaultValue()); >> >> statAccumulator.value().setYearMetric(yearTotal); >> >> >> >> >> >> >> >> } >> >> } >> > >> > >> > >> >-- >> >Best, >> >Hangxiang. >>