Hello everyone. I'm trying to play with

MapState

in PyFlink and have the following code:

...

class DelaySumFunction(KeyedProcessFunction):
    def __init__(self):
        self.map_state = None

    def open(self, runtime_context):
        state_ttl_config = (
            StateTtlConfig.new_builder(Time.minutes(5))
            .set_state_visibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
            .build()
        )

        state_descriptor = MapStateDescriptor("delays", <http://Types.INT|Types.INT>(), <http://Types.INT|Types.INT>())
        state_descriptor.enable_time_to_live(state_ttl_config)

        self.map_state = runtime_context.get_state(state_descriptor)

    def process_element(self, value, ctx: "ProcessFunction.Context"):
        self.map_state.put(value[0], value[1])

        yield Row(self.map_state.values().sum())

...

When I run pipeline with this process function I get the following exception and cannot figure it out

File "/opt/flink/usrlib/prague-sum-delays.py", line 31, in process_element
    self.map_state.put(value[0], value[1])
AttributeError: 'SynchronousValueRuntimeState' object has no attribute 'put'

Anyone have any pointers?

what exactly you are trying to do?

@Siddhesh Kalgaonkar I have a stream of events with position of tram vehicles and their current delay. In this exercise I'm trying to get a sum of delays of all currently running vehicles. When a new event from a specific vehicle comes in I need to reflect it's updated delay (could be smaller or larger) and update the sum accordingly.