Python statefun - Context update

[Jérémy Albrecht <ja...@skapane.ai>]

Hi all !

I am currently developing an environment that uses Flink Stateful functions in Python. The architecture is complex but here are the main points that needs to be understood to frame the problem I am facing.
* functions.py contains several methods, one of them is handling protobuf messages and storing into context.storage a ValueSpec which is a python Object. This object contains classical types such as natives types, but also lists of named tuple and objects.
Since pickle is not able to serialize objects and functions, I use dill to serialize this object and make a Flink-compatible type. 
* The object stored in the Context can be resumed as this:
@dataclass
class RunState:
    duration: int
    runner: Runner
and runner is also an instance of the class Runner, defined as:
class Runner:
    counter: int = 0
    
    def prepare_smth(self):
        self.counter = 10
        
        def add_to_counter():
            self.counter += 1
       
       return add_to_counter

From the object stored in the state of the Flink function I so have access to the function definition *add_to_counter* and I can do something like:
method_caller = ctx.storage.run_state.runner.prepare_smth()
method_caller()

-> from now on I expect the attribute counter from the Runner instance defined as runner in the Run State to have the value 11. What happens is that the value is modified inside the context of add_to_counter and prepare_smth, but the change is never reflected inside the object stored in the context. ctx.storage.state.runner.counter still equals to 0. 

* After some research with the debugger, in my opinion, the program tries to update the value but ctx.storage seems to be only updatable when I do a reassignement: 
e.g. ctx.storage.run_state.counter += 1 has NO effect, but 
rs = ctx.storage.run_state
rs.counter +=1
ctx.storage.run_state = rs has the expected result.

If you have any clue about how to start or what to do about my problem I greatly appreciate the help !!
Don't hesitate to ask me if anything is unclear :)

Thanks,
Jérémy
    

Re: Python statefun - Context update

[Nicolaus Weidner <ni...@ververica.com>]

Hi Jérémy,

objects are serialized when you store them in state. So when you retrieve
run_state from state, it is deserialized and you have a fresh instance.
Calling method_caller() then modifies this instance, but *not *the
serialized version stored in state.
In the second attempt you described, you modified the retrieved instance,
then stored the modified version in state (or rather a serialized form of
it) - which works as expected.

I don't think there is any way around explicitly storing any local changes
you made in state.

Best,
Nico

On Mon, Sep 20, 2021 at 5:22 PM Jérémy Albrecht <ja...@skapane.ai>
wrote:

> Hi all !
>
> I am currently developing an environment that uses Flink Stateful
> functions in Python. The architecture is complex but here are the main
> points that needs to be understood to frame the problem I am facing.
> * functions.py contains several methods, one of them is handling protobuf
> messages and storing into context.storage a ValueSpec which is a python
> Object. This object contains classical types such as natives types, but
> also lists of named tuple and objects.
> Since pickle is not able to serialize objects and functions, I use dill to
> serialize this object and make a Flink-compatible type.
> * The object stored in the Context can be resumed as this:
> @dataclass
> class RunState:
>     duration: int
>     runner: Runner
> and runner is also an instance of the class Runner, defined as:
> class Runner:
>     counter: int = 0
>
>     def prepare_smth(self):
>         self.counter = 10
>
>         def add_to_counter():
>             self.counter += 1
>
>        return add_to_counter
>
> From the object stored in the state of the Flink function I so have access
> to the function definition *add_to_counter* and I can do something like:
> method_caller = ctx.storage.run_state.runner.prepare_smth()
> method_caller()
>
> -> from now on I expect the attribute counter from the Runner instance
> defined as runner in the Run State to have the value 11. What happens is
> that the value is modified inside the context of add_to_counter and
> prepare_smth, but the change is never reflected inside the object stored in
> the context. ctx.storage.state.runner.counter still equals to 0.
>
> * After some research with the debugger, in my opinion, the program tries
> to update the value but ctx.storage seems to be only updatable when I do a
> reassignement:
> e.g. ctx.storage.run_state.counter += 1 has NO effect, but
> rs = ctx.storage.run_state
> rs.counter +=1
> ctx.storage.run_state = rs has the expected result.
>
> If you have any clue about how to start or what to do about my problem I
> greatly appreciate the help !!
> Don't hesitate to ask me if anything is unclear :)
>
> Thanks,
> Jérémy
>
>