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
> > > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>
