Partitioning operator state

[Dominik Safaric]

Hi everyone,

In the case of scaling out a Flink cluster, how does Flink handle operator state partitioning of a staged topology?

Regards,
Dominik

[Tzu-Li (Gordon) Tai]

Hi Dominik,

Do you mean how Flink redistributes an operator’s state when the parallelism of the operator is changed?

If so, you can take a look at [1] and [2].

Cheers,

Gordon

[1] https://issues.apache.org/jira/browse/FLINK-3755

[2] https://docs.google.com/document/d/1G1OS1z3xEBOrYD4wSu-LuBCyPUWyFd9l3T9WyssQ63w/edit#

On December 8, 2016 at 4:40:18 AM, Dominik Safaric ([hidden email]) wrote:

Hi everyone,

In the case of scaling out a Flink cluster, how does Flink handle operator state partitioning of a staged topology?

Regards,
Dominik

[Stefan Richter]

Hi Dominik,

as Gordon’s response only covers keyed-state, I will briefly explain what happens for non-keyed operator state. In contrast to Flink 1.1, Flink 1.2 checkpointing does not write a single blackbox object (e.g. ONE object that is a set of all kafka offsets is emitted), but a list of blackbox objects instead (e.g. think of all kafka offsets being emitted individually, as MULTIPLE objects). While Flink 1.2 still has no knowledge about the emitted objects in the list (thus they remain blackboxes), what the contract allows is that those objects can be freely redistributed in case of scale-out or scale-in. Scaling is merely splitting or merging of the checkpointed lists.

Best,

Stefan 

Am 08.12.2016 um 08:00 schrieb Tzu-Li (Gordon) Tai <[hidden email]>:

Hi Dominik,

Do you mean how Flink redistributes an operator’s state when the parallelism of the operator is changed?

If so, you can take a look at [1] and [2].

Cheers,

Gordon

[1] https://issues.apache.org/jira/browse/FLINK-3755

[2] https://docs.google.com/document/d/1G1OS1z3xEBOrYD4wSu-LuBCyPUWyFd9l3T9WyssQ63w/edit#

On December 8, 2016 at 4:40:18 AM, Dominik Safaric ([hidden email]) wrote:

Hi everyone, 

In the case of scaling out a Flink cluster, how does Flink handle operator state partitioning of a staged topology?  

Regards, 
Dominik 

[Dominik Safaric]

Dear Stefan, 

Thanks for the clarification. 

How is however the state recovered in the case of a task failure? Assuming there is a topology of 10 workers and a worker dies. The state in this case, after restarting the entire execution, will how exactly be distributed across the workers? 

Dominik

On 8 Dec 2016, at 10:04, Stefan Richter <[hidden email]> wrote:

Hi Dominik,

as Gordon’s response only covers keyed-state, I will briefly explain what happens for non-keyed operator state. In contrast to Flink 1.1, Flink 1.2 checkpointing does not write a single blackbox object (e.g. ONE object that is a set of all kafka offsets is emitted), but a list of blackbox objects instead (e.g. think of all kafka offsets being emitted individually, as MULTIPLE objects). While Flink 1.2 still has no knowledge about the emitted objects in the list (thus they remain blackboxes), what the contract allows is that those objects can be freely redistributed in case of scale-out or scale-in. Scaling is merely splitting or merging of the checkpointed lists.

Best,

Stefan 

Am 08.12.2016 um 08:00 schrieb Tzu-Li (Gordon) Tai <[hidden email]>:

Hi Dominik,

Do you mean how Flink redistributes an operator’s state when the parallelism of the operator is changed?

If so, you can take a look at [1] and [2].

Cheers,

Gordon

[1] https://issues.apache.org/jira/browse/FLINK-3755

[2] https://docs.google.com/document/d/1G1OS1z3xEBOrYD4wSu-LuBCyPUWyFd9l3T9WyssQ63w/edit#

On December 8, 2016 at 4:40:18 AM, Dominik Safaric ([hidden email]) wrote:

Hi everyone, 

In the case of scaling out a Flink cluster, how does Flink handle operator state partitioning of a staged topology?  

Regards, 
Dominik