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