HI David Thanks a lot. I almost get the point. When I use initializeState to restore the mapstate, the task can not get a key at that moment, so I just get the key but not the UK, when I use the mapstate in processElement, a key will be provided implictly, so I would get the right UK and UV. But still I think I should get <key,<UK,UV>> at the initializeState but not the <key, UV>. Any way, I have changed my code to just use the mapstate provided by flink. Thanks.
Yours sincerely Josh David Morávek <[email protected]> 于2021年12月29日周三 23:01写道： > The problem is that you're not actually using the underlying state during > runtime, but instead you're simply using a java map abstraction. This > property ("Map<Long, SubState> state") is simply bound to the UDF lifecycle > and doesn't share the semantics of the keyed state. > > You should be using the "MapState" property directly to get the guarantees > you're looking for. Then you also won't need to override the snapshot / > initialize state methods, which simplifies the code a lot. > > D. > > On Wed, Dec 29, 2021 at 2:08 PM Joshua Fan <[email protected]> wrote: > >> Hi David, >> Thanks for you reply. >> Yes, for keyed state, every state is referenced by a particular key, but >> I would guess it is a flink sdk issue, I mean, the keyed state maybe saved >> as (key, keyed state), as for my situation, it is (key, mapstate(UK,UV)), >> I think the key of this pair is not easy to get by user, when I do >> mapstate.keyset I want to get the UK set, not the key set. According to my >> job, the (key, mapstate(UK,UV)) can be get successfully when job is >> running, but when job restarts from a checkpoint, the restored mapstate, >> the pair seemed be changed to (key, UV), the UK just gone, I can not find >> back the UK. I think the key of (key, mapstate(UK,UV)) will be implictly >> added when write or read from the state by flink. >> So, I am still not clear why I get the key but not the UK. >> >> Yours >> Josh >> >> David Morávek <[email protected]> 于2021年12月29日周三 17:32写道： >> >>> Hi Josh, >>> >>> it's important bit to understand is that the MapState (or any other >>> keyed state) is scoped per *key* [1]. You can think about it in a way, >>> that for each key you have a separate "map" that backs it. This is the >>> important concept behind distributed stream processing, that allows you to >>> parallelize the computation and still make sure, that all data for the same >>> key end up in the same partition. >>> >>> Does this answer your question? >>> >>> [1] >>> https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/concepts/stateful-stream-processing/#keyed-state >>> >>> Best, >>> D. >>> >>
