Blocking or pipelined mode for batch job

[Zhijiang(wangzhijiang999)]

Hi ,

       As I reviewed the flink source code, if the ExecutionMode is set "Pipelined", the DataExechangeMode will be set "Batch" if breaking pipelined property is true for two input or iteration situation in order to avoid deadlock. When the DataExechangeMode is set "Batch", the ResultPartitionType will be set "Pipelined" if dynamic path for iteration . So the final mode for dynamic node in iteration is "Pipelined"? My understanding is right ? If it is , why must use pipelined mode for dynamic node in iteration?

    Another question is that when running a batch job, if one task failed, do i must restart the whole topology or are there any machnism to just restart failed task?  Thank you for any advise !

[Ufuk Celebi]

On Tue, May 10, 2016 at 10:56 AM, wangzhijiang999
<[hidden email]> wrote:
>        As I reviewed the flink source code, if the ExecutionMode is set
> "Pipelined", the DataExechangeMode will be set "Batch" if breaking pipelined
> property is true for two input or iteration situation in order to avoid
> deadlock. When the DataExechangeMode is set "Batch", the ResultPartitionType
> will be set "Pipelined" if dynamic path for iteration . So the final mode
> for dynamic node in iteration is "Pipelined"? My understanding is right ? If
> it is , why must use pipelined mode for dynamic node in iteration?

Yes, this is a current shortcoming, because the blocking results can
not be used within iterations (they can not be reset after each
superstep).

>     Another question is that when running a batch job, if one task failed,
> do i must restart the whole topology or are there any machnism to just
> restart failed task?  Thank you for any advise !

Currently, yes, the whole topology is restarted. There was an attempt
to add support for more fine-grained fault tolerance, but it was never
completed/merged.

– Ufuk

[Zhijiang(wangzhijiang999)]

Hi Ufuk, 

     Thank you for quick response! I am not very clear of the internal realize for iteration, so would you explain in detail why blocking results can not be reset after each superstep?

In addition,  for the below example, why it may cause deadlock in pipelined mode? 

DataSet mapped1 = data.map(new MyMapper());
DataSet mapped2 = data.map(new AnotherMapper());
mapped1.join(mapped2).where(...).equalTo(...);

------------------------------------------------------------------

发件人：Ufuk Celebi <[hidden email]>

发送时间：2016年5月10日(星期二) 17:22

收件人：user <[hidden email]>; wangzhijiang999 <[hidden email]>

主　题：Re: Blocking or pipelined mode for batch job

On Tue, May 10, 2016 at 10:56 AM, wangzhijiang999
<[hidden email]> wrote:
>        As I reviewed the flink source code, if the ExecutionMode is set
> "Pipelined", the DataExechangeMode will be set "Batch" if breaking pipelined
> property is true for two input or iteration situation in order to avoid
> deadlock. When the DataExechangeMode is set "Batch", the ResultPartitionType
> will be set "Pipelined" if dynamic path for iteration . So the final mode
> for dynamic node in iteration is "Pipelined"? My understanding is right ? If
> it is , why must use pipelined mode for dynamic node in iteration?

Yes, this is a current shortcoming, because the blocking results can
not be used within iterations (they can not be reset after each
superstep).

>     Another question is that when running a batch job, if one task failed,
> do i must restart the whole topology or are there any machnism to just
> restart failed task?  Thank you for any advise !

Currently, yes, the whole topology is restarted. There was an attempt
to add support for more fine-grained fault tolerance, but it was never
completed/merged.

– Ufuk