Hi All My flink version is 1.11, the statebackend is rocksdb, and I want to write a flink job to implement an adaptive window. I wrote a flink dag like below:
> DataStream<DataEntity> entities = env.addSource(new > EntitySource()).setParallelism(1); > > entities.keyBy(DataEntity::getName).process(new > EntityKeyedProcessFunction()).setParallelism(p); > > The important code is the EntityKeyedProcessFunction, it is attached. I have a mapstate in it like 'private transient MapState<String, SubState> entityStates;' I print the content of the mapstate when checkpoint completed, the content is ok like below: > 2021-12-28 16:22:05,487 INFO window.EntityKeyedProcessFunction [] - > >>>>>>>>>the key is 1640679600000 > 2021-12-28 16:22:05,487 INFO window.EntityKeyedProcessFunction [] - the > value is name = hippopotamus, value = [window.DataEntity@b9266a0, window. > DataEntity@682fce90] > 2021-12-28 16:22:05,488 INFO window.EntityKeyedProcessFunction [] - the > value is name = crocodile, value = [window.DataEntity@16d20045] > 2021-12-28 16:22:05,488 INFO window.EntityKeyedProcessFunction [] - the > value is name = dolphin, value = [window.DataEntity@1820c75a, window. > DataEntity@64f3b9f6] > 2021-12-28 16:22:05,488 INFO window.EntityKeyedProcessFunction [] - the > value is name = Dragonfly, value = [window.DataEntity@2b2ad03] > 2021-12-28 16:22:05,488 INFO window.EntityKeyedProcessFunction [] - the > value is name = Hedgehog, value = [window.DataEntity@65f39671, window. > DataEntity@2df6b2bf] > 2021-12-28 16:22:05,488 INFO window.EntityKeyedProcessFunction [] - the > value is name = Bee, value = [window.DataEntity@13249998] > 2021-12-28 16:22:05,488 INFO window.EntityKeyedProcessFunction [] - the > value is name = cicada, value = [window.DataEntity@7266e125, window. > DataEntity@167cf1ae] > 2021-12-28 16:22:05,488 INFO window.EntityKeyedProcessFunction [] - the > value is name = Mosquito, value = [window.DataEntity@2596aa5a, window. > DataEntity@603c0804] > 2021-12-28 16:22:05,488 INFO window.EntityKeyedProcessFunction [] - the > value is name = Crane, value = [window.DataEntity@2a3192e9, window. > DataEntity@3a65398f] > The key of the mapstate is a time which was coalesced to minute. But when the job restarted from a checkpoint, the content of the mapstate changed, actually, the key of the mapstate changed. It would show as below. > 2021-12-28 16:15:45,379 INFO window.EntityKeyedProcessFunction [] - > >>>>>>>>>the key is carp > 2021-12-28 16:15:45,391 INFO window.EntityKeyedProcessFunction [] - the > value is name = hippopotamus, value = [window.DataEntity@510d4c4b, window. > DataEntity@7857e387] > 2021-12-28 16:15:45,391 INFO window.EntityKeyedProcessFunction [] - the > value is name = crocodile, value = [window.DataEntity@31366a33, window. > DataEntity@a62074a] > 2021-12-28 16:15:45,391 INFO window.EntityKeyedProcessFunction [] - the > value is name = Dragonfly, value = [window.DataEntity@56db63fa, window. > DataEntity@54befce0, window.DataEntity@4e7cf96a] > 2021-12-28 16:15:45,391 INFO window.EntityKeyedProcessFunction [] - the > value is name = Hedgehog, value = [window.DataEntity@7ad09313, window. > DataEntity@592a2955] > 2021-12-28 16:15:45,391 INFO window.EntityKeyedProcessFunction [] - the > value is name = Mosquito, value = [window.DataEntity@48c05cae] > 2021-12-28 16:15:45,392 INFO window.EntityKeyedProcessFunction [] - the > value is name = duck, value = [window.DataEntity@3e9ef1a4] > 2021-12-28 16:15:45,392 INFO window.EntityKeyedProcessFunction [] - the > value is name = Rhinoceros, value = [window.DataEntity@25f11701, window. > DataEntity@2334b667] > 2021-12-28 16:15:45,392 INFO window.EntityKeyedProcessFunction [] - the > value is name = Eagle, value = [window.DataEntity@7574eb4a] > It seems like the key of the restored mapstate is the key of the operator. My minute time was gone, and it is replaced by the key of the operator. It is so weird. Do I misuse the mapstate? Thanks. Yours Josh
EntityKeyedProcessFunction.java
Description: Binary data
