Related to the above: Why would I be able to access

MapState.items()

from my

process_element

function but get the following error when calling it from

open

TypeError: None has type NoneType, but expected one of: bytes, unicode

Does that just mean the state hasn't actually been created yet?

Are you saying you cannot access state in the

open

method? This is by design. Each state is scoped under a specific key which is known only when processing elements. For the following job:

class MyProcessFunction(KeyedProcessFunction):
    def open(self, runtime_context: RuntimeContext):
        self.map_state = runtime_context.get_map_state(
            MapStateDescriptor("map", Types.STRING(), <http://Types.INT|Types.INT>()))

    def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
         pass
      

ds = env.from_collection(
    collection=[(1, 'a'), (1, 'b'), (2, 'c')],
    type_info=Types.ROW([<http://Types.INT|Types.INT>(), Types.STRING()]))
keyed_ds = ds.key_by(lambda a: a[0])
keyed_ds.process

For example, for

map_state

, its structure is actually

Map<INT, Map<STRING, INT>>

(suppose named it as my_map) where the outer map is hiden from users. For each input elements: • (1, ‘a’), the key is 1, and so the map state you get in process_element is actually my_map.get(1) • (1, ‘b’), the key is 1, and so the map state you get in process_element is actually my_map.get(1) • (2, ‘c’), the key is 2, and so the map state you get in process_element is actually my_map.get(2) In this way, for each input elements, you could only get the state scoped under the same key. This is by design in Flink. Regarding

open

, there are still no input elements and so the current key is still not set. So you cannot access it.

Yeah, I'm just surprised I can

get_state

and

get_map_state

in

open

, but I can't use them.

@Nathanael England Oh, I sent out an incomplete reply for mistake. Please take a look at if the above reply could answer your questions.

open

isn't called until there are elements so that was making me think it did know the scope at the time of running that method.

open

is called during initialization. Per my understanding, it should be called before there are any elements.

Yup, I think that was the piece I was missing in my original investigation. The API just left a cryptic error, but I think I have the same understanding as you! Thanks for helping me confirm this behavior.