In general, I agree with you about aiming jobs with no/few blocking
exchanges for fine-grained recovery. The only problem is, correct me
if I'm wrong, users currently cannot control the data exchanging mode
of a specific edge. I'm not aware of such APIs.
As a first step, I'd prefer excluding this from the scope of this FLIP.
Best,
Xintong
On Wed, May 25, 2022 at 8:54 PM Chesnay Schepler <ches...@apache.org>
wrote:
Yes; but that's also a limitation of the current fine-grained
recovery.
My suggestion was primarily aimed at jobs that have no/few blocking
exchanges, where users would currently have to explicitly configure
additional blocking exchanges to really get something out of
fine-grained recovery (at the expense of e2e job duration).
On 25/05/2022 14:47, Xintong Song wrote:
>> Will this also allow spilling everything to disk while also
forwarding
>> data to the next task?
>>
> Yes, as long as the downstream task is started, this always
forward the
> data, even while spilling everything.
>
> This would allow us to improve fine-grained recovery by no
longer being
>> constrained to pipelined regions.
>
> I think it helps preventing restarts of the upstreams for a
failed task,
> but not the downstreams. Because there's no guarantee a
restarted task will
> prevent exactly same data (in terms of order) as the previous
execution,
> thus downstreams cannot resume consuming the data.
>
>
> Best,
>
> Xintong
>
>
>
> On Wed, May 25, 2022 at 3:05 PM Chesnay Schepler
<ches...@apache.org> wrote:
>
>> Will this also allow spilling everything to disk while also
forwarding
>> data to the next task?
>>
>> This would allow us to improve fine-grained recovery by no
longer being
>> constrained to pipelined regions.
>>
>> On 25/05/2022 05:55, weijie guo wrote:
>>> Hi All,
>>> Thank you for your attention and feedback.
>>> Do you have any other comments? If there are no other
questions, I'll
>> vote
>>> on FLIP-235 tomorrow.
>>>
>>> Best regards,
>>>
>>> Weijie
>>>
>>>
>>> Aitozi <gjying1...@gmail.com> 于2022年5月20日周五 13:22写道：
>>>
>>>> Hi Xintong
>>>> Thanks for your detailed explanation, I misunderstand
the spill
>>>> behavior at first glance,
>>>> I get your point now. I think it will be a good addition to
the current
>>>> execution mode.
>>>> Looking forward to it :)
>>>>
>>>> Best,
>>>> Aitozi
>>>>
>>>> Xintong Song <tonysong...@gmail.com> 于2022年5月20日周五
12:26写道：
>>>>
>>>>> Hi Aitozi,
>>>>>
>>>>> In which case we can use the hybrid shuffle mode
>>>>>
>>>>> Just to directly answer this question, in addition to
>>>>> Weijie's explanations. For batch workload, if you want the
workload to
>>>> take
>>>>> advantage of as many resources as available, which ranges
from a single
>>>>> slot to as many slots as the total tasks, you may consider
hybrid
>> shuffle
>>>>> mode. Admittedly, this may not always be wanted, e.g., users
may not
>> want
>>>>> to execute a job if there's too few resources available, or
may not
>> want
>>>> a
>>>>> job taking too many of the cluster resources. That's why we
propose
>>>> hybrid
>>>>> shuffle as an additional option for batch users, rather than a
>>>> replacement
>>>>> for Pipelined or Blocking mode.
>>>>>
>>>>> So you mean the hybrid shuffle mode will limit its usage to
the bounded
>>>>>> source, Right ?
>>>>>>
>>>>> Yes.
>>>>>
>>>>> One more question, with the bounded data and partly of the
stage is
>>>> running
>>>>>> in the Pipelined shuffle mode, what will be the behavior of
the task
>>>>>> failure, Is the checkpoint enabled for these running stages
or will it
>>>>>> re-run after the failure?
>>>>>>
>>>>> There's no checkpoints. The failover behavior depends on the
spilling
>>>>> strategy.
>>>>> - In the first version, we only consider a selective
spilling strategy,
>>>>> which means spill data as little as possible to the disk,
which means
>> in
>>>>> case of failover upstream tasks need to be restarted to
reproduce the
>>>>> complete intermediate results.
>>>>> - An alternative strategy we may introduce in future if
needed is to
>>>> spill
>>>>> the complete intermediate results. That avoids restarting
upstream
>> tasks
>>>> in
>>>>> case of failover, because the produced intermediate results
can be
>>>>> re-consumed, at the cost of more disk IO load.
>>>>> With both strategies, the trade-off between failover cost
and IO load
>> is
>>>>> for the user to decide. This is also discussed in the
MemoryDataManager
>>>>> section of the FLIP.
>>>>>
>>>>> Best,
>>>>>
>>>>> Xintong
>>>>>
>>>>>
>>>>>
>>>>> On Fri, May 20, 2022 at 12:10 PM Aitozi
<gjying1...@gmail.com> wrote:
>>>>>
>>>>>> Thanks Weijie for your answer. So you mean the hybrid
shuffle mode
>> will
>>>>>> limit
>>>>>> its usage to the bounded source, Right ?
>>>>>> One more question, with the bounded data and partly of the
stage is
>>>>> running
>>>>>> in the Pipelined shuffle mode, what will be the behavior of
the task
>>>>>> failure, Is the
>>>>>> checkpoint enabled for these running stages or will it
re-run after
>> the
>>>>>> failure?
>>>>>>
>>>>>> Best,
>>>>>> Aitozi
>>>>>>
>>>>>> weijie guo <guoweijieres...@gmail.com> 于2022年5月20日周五
10:45写道：
>>>>>>
>>>>>>> Hi, Aitozi:
>>>>>>>
>>>>>>> Thank you for the feedback!
>>>>>>> Here are some of my thoughts on your question
>>>>>>>
>>>>>>>>>> 1.If there is an unbounded data source, but only have
resource to
>>>>>>> schedule the first stage, will it bring the big burden to the
>>>>>> disk/shuffle
>>>>>>> service which will occupy all the resource I think.
>>>>>>> First of all, Hybrid Shuffle Mode is oriented to the batch job
>>>>> scenario,
>>>>>> so
>>>>>>> there is no problem of unbounded data sources. Secondly,
if you
>>>>> consider
>>>>>>> the stream scenario, I think Pipelined Shuffle should
still be the
>>>> best
>>>>>>> choice at present. For an unbounded data stream, it is not
meaningful
>>>>> to
>>>>>>> only run some stages.
>>>>>>>
>>>>>>>>>> 2. Which kind of job will benefit from the hybrid
shuffle mode.
>>>> In
>>>>>>> other words, In which case we can use the hybrid shuffle mode:
>>>>>>> Both general batch jobs and OLAP jobs benefit. For batch
jobs, hybrid
>>>>>>> shuffle mode can effectively utilize cluster resources and
avoid some
>>>>>>> unnecessary disk IO overhead. For OLAP scenarios, which are
>>>>> characterized
>>>>>>> by a large number of concurrently submitted short batch
jobs, hybrid
>>>>>>> shuffle can solve the scheduling deadlock problem of pipelined
>>>> shuffle
>>>>>> and
>>>>>>> achieve similar performance.
>>>>>>>
>>>>>>> Best regards,
>>>>>>>
>>>>>>> Weijie
>>>>>>>
>>>>>>>
>>>>>>> Aitozi <gjying1...@gmail.com> 于2022年5月20日周五 08:05写道：
>>>>>>>
>>>>>>>> Hi Weijie:
>>>>>>>>
>>>>>>>> Thanks for the nice FLIP, I have couple questions
about this:
>>>>>>>>
>>>>>>>> 1) In the hybrid shuffle mode, the shuffle mode is
decided by the
>>>>>>> resource.
>>>>>>>> If there
>>>>>>>> is an unbounded data source, but only have resource to
schedule the
>>>>>> first
>>>>>>>> stage, will it
>>>>>>>> bring the big burden to the disk/shuffle service which
will occupy
>>>>> all
>>>>>>> the
>>>>>>>> resource I think.
>>>>>>>>
>>>>>>>> 2) Which kind of job will benefit from the hybrid shuffle
mode. In
>>>>>> other
>>>>>>>> words, In which
>>>>>>>> case we can use the hybrid shuffle mode:
>>>>>>>> - For batch job want to use more resource to reduce the
e2e time ?
>>>>>>>> - Or for streaming job which may lack of resource
temporarily ?
>>>>>>>> - Or for OLAP job which will try to make best use of
available
>>>>>> resources
>>>>>>> as
>>>>>>>> you mentioned to finish the query?
>>>>>>>> Just want to know the typical use case for the Hybrid
shuffle mode
>>>> :)
>>>>>>>> Best,
>>>>>>>> Aitozi.
>>>>>>>>
>>>>>>>> weijie guo <guoweijieres...@gmail.com> 于2022年5月19日周四
18:33写道：
>>>>>>>>
>>>>>>>>> Yangze, Thank you for the feedback!
>>>>>>>>> Here's my thoughts for your questions:
>>>>>>>>>
>>>>>>>>>>>> How do we decide the size of the buffer pool in
>>>>> MemoryDataManager
>>>>>>> and
>>>>>>>>> the read buffers in FileDataManager?
>>>>>>>>> The BufferPool in MemoryDataManager is the
LocalBufferPool used
>>>> by
>>>>>>>>> ResultPartition, and the size is the same as the current
>>>>>> implementation
>>>>>>>> of
>>>>>>>>> sort-merge shuffle. In other words, the minimum value of
>>>> BufferPool
>>>>>> is
>>>>>>> a
>>>>>>>>> configurable fixed value, and the maximum value is
Math.max(min,
>>>> 4
>>>>> *
>>>>>>>>> numSubpartitions). The default value can be determined
by running
>>>>> the
>>>>>>>>> TPC-DS tests.
>>>>>>>>> Read buffers in FileDataManager are requested from the
>>>>>>>>> BatchShuffleReadBufferPool shared by TaskManager, it's size
>>>>>> controlled
>>>>>>> by
>>>>>>>>>
*taskmanager.memory.framework.off-heap.batch-shuffle.size*, the
>>>>>> default
>>>>>>>>> value is 32M, which is consistent with the current
sort-merge
>>>>> shuffle
>>>>>>>>> logic.
>>>>>>>>>
>>>>>>>>>>>> Is there an upper limit for the sum of them? If there
is, how
>>>>>> does
>>>>>>>>> MemoryDataManager and FileDataManager sync the memory usage?
>>>>>>>>> The buffers of the MemoryDataManager are limited by the
size of
>>>> the
>>>>>>>>> LocalBufferPool, and the upper limit is the size of the
Network
>>>>>> Memory.
>>>>>>>> The
>>>>>>>>> buffers of the FileDataManager are directly requested from
>>>>>>>>> UnpooledOffHeapMemory, and are also limited by the size
of the
>>>>>>> framework
>>>>>>>>> off-heap memory. I think there should be no need for
additional
>>>>>>>>> synchronization mechanisms.
>>>>>>>>>
>>>>>>>>>>>> How do you disable the slot sharing? If user
configures both
>>>>> the
>>>>>>> slot
>>>>>>>>> sharing group and hybrid shuffle, what will happen to
that job?
>>>>>>>>> I think we can print a warning log when Hybrid Shuffle
is enabled
>>>>> and
>>>>>>> SSG
>>>>>>>>> is configured during the JobGraph compilation stage, and
fallback
>>>>> to
>>>>>>> the
>>>>>>>>> region slot sharing group by default. Of course, it will be
>>>>>> emphasized
>>>>>>> in
>>>>>>>>> the document that we do not currently support SSG, If
configured,
>>>>> it
>>>>>>> will
>>>>>>>>> fall back to the default.
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> Best regards,
>>>>>>>>>
>>>>>>>>> Weijie
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> Yangze Guo <karma...@gmail.com> 于2022年5月19日周四
16:25写道：
>>>>>>>>>
>>>>>>>>>> Thanks for driving this. Xintong and Weijie.
>>>>>>>>>>
>>>>>>>>>> I believe this feature will make Flink a better batch/OLAP
>>>>> engine.
>>>>>> +1
>>>>>>>>>> for the overall design.
>>>>>>>>>>
>>>>>>>>>> Some questions:
>>>>>>>>>> 1. How do we decide the size of the buffer pool in
>>>>>> MemoryDataManager
>>>>>>>>>> and the read buffers in FileDataManager?
>>>>>>>>>> 2. Is there an upper limit for the sum of them? If
there is,
>>>> how
>>>>>> does
>>>>>>>>>> MemoryDataManager and FileDataManager sync the memory
usage?
>>>>>>>>>> 3. How do you disable the slot sharing? If user
configures both
>>>>> the
>>>>>>>>>> slot sharing group and hybrid shuffle, what will happen
to that
>>>>>> job?
>>>>>>>>>> Best,
>>>>>>>>>> Yangze Guo
>>>>>>>>>>
>>>>>>>>>> On Thu, May 19, 2022 at 2:41 PM Xintong Song <
>>>>>> tonysong...@gmail.com>
>>>>>>>>>> wrote:
>>>>>>>>>>> Thanks for preparing this FLIP, Weijie.
>>>>>>>>>>>
>>>>>>>>>>> I think this is a good improvement on batch resource
>>>>> elasticity.
>>>>>>>>> Looking
>>>>>>>>>>> forward to the community feedback.
>>>>>>>>>>>
>>>>>>>>>>> Best,
>>>>>>>>>>>
>>>>>>>>>>> Xintong
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> On Thu, May 19, 2022 at 2:31 PM weijie guo <
>>>>>>>> guoweijieres...@gmail.com>
>>>>>>>>>>> wrote:
>>>>>>>>>>>
>>>>>>>>>>>> Hi all,
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> I’d like to start a discussion about FLIP-235[1], which
>>>>>>> introduce a
>>>>>>>>>> new shuffle mode
>>>>>>>>>>>> can overcome some of the problems of Pipelined
Shuffle and
>>>>>>>> Blocking
>>>>>>>>>> Shuffle in batch scenarios.
>>>>>>>>>>>> Currently in Flink, task scheduling is more or less
>>>>> constrained
>>>>>>> by
>>>>>>>>> the
>>>>>>>>>> shuffle implementations.
>>>>>>>>>>>> This will bring the following disadvantages:
>>>>>>>>>>>>
>>>>>>>>>>>> 1. Pipelined Shuffle:
>>>>>>>>>>>> For pipelined shuffle, the upstream and downstream
>>>> tasks
>>>>>> are
>>>>>>>>>> required to be deployed at the same time, to avoid upstream
>>>> tasks
>>>>>>> being
>>>>>>>>>> blocked forever. This is fine when there are enough
resources
>>>> for
>>>>>>> both
>>>>>>>>>> upstream and downstream tasks to run simultaneously,
but will
>>>>> cause
>>>>>>> the
>>>>>>>>>> following problems otherwise:
>>>>>>>>>>>> 1.
>>>>>>>>>>>> Pipelined shuffle connected tasks (i.e., a pipelined
>>>>>>> region)
>>>>>>>>>> cannot be executed until obtaining resources for all of
them,
>>>>>>> resulting
>>>>>>>>> in
>>>>>>>>>> longer job finishing time and poorer resource
efficiency due to
>>>>>>> holding
>>>>>>>>>> part of the resources idle while waiting for the rest.
>>>>>>>>>>>> 2.
>>>>>>>>>>>> More severely, if multiple jobs each hold
part of the
>>>>>>> cluster
>>>>>>>>>> resources and are waiting for more, a deadlock would
occur. The
>>>>>>> chance
>>>>>>>> is
>>>>>>>>>> not trivial, especially for scenarios such as OLAP where
>>>>> concurrent
>>>>>>> job
>>>>>>>>>> submissions are frequent.
>>>>>>>>>>>> 2. Blocking Shuffle:
>>>>>>>>>>>> For blocking shuffle, execution of downstream tasks
>>>> must
>>>>>> wait
>>>>>>>> for
>>>>>>>>>> all upstream tasks to finish, despite there might be more
>>>>> resources
>>>>>>>>>> available. The sequential execution of upstream and
downstream
>>>>>> tasks
>>>>>>>>>> significantly increase the job finishing time, and the
disk IO
>>>>>>> workload
>>>>>>>>> for
>>>>>>>>>> spilling and loading full intermediate data also
affects the
>>>>>>>> performance.
>>>>>>>>>>>> We believe the root cause of the above problems is that
>>>>> shuffle
>>>>>>>>>> implementations put unnecessary constraints on task
scheduling.
>>>>>>>>>>>> To solve this problem, Xintong Song and I propose to
>>>>> introduce
>>>>>>>> hybrid
>>>>>>>>>> shuffle to minimize the scheduling constraints. With Hybrid
>>>>>> Shuffle,
>>>>>>>>> Flink
>>>>>>>>>> should:
>>>>>>>>>>>> 1. Make best use of available resources.
>>>>>>>>>>>> Ideally, we want Flink to always make progress if
>>>>> possible.
>>>>>>>> That
>>>>>>>>>> is to say, it should always execute a pending task if
there are
>>>>>>>> resources
>>>>>>>>>> available for that task.
>>>>>>>>>>>> 2. Minimize disk IO load.
>>>>>>>>>>>> In-flight data should be consumed directly from memory
>>>> as
>>>>>>> much
>>>>>>>> as
>>>>>>>>>> possible. Only data that is not consumed timely should be
>>>> spilled
>>>>>> to
>>>>>>>>> disk.
>>>>>>>>>>>> You can find more details in FLIP-235. Looking forward to
>>>>> your
>>>>>>>>>> feedback.
>>>>>>>>>>>> [1]
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>>
>>
https://cwiki.apache.org/confluence/display/FLINK/FLIP-235%3A+Hybrid+Shuffle+Mode
>>>>>>>>>>>>
>>>>>>>>>>>> Best regards,
>>>>>>>>>>>>
>>>>>>>>>>>> Weijie
>>>>>>>>>>>>
>>
