Thanks for the responses, Till and Xintong. I second Xintong's comment that SSG-based runtime interface will give us the flexibility to achieve op/task-based approach. That's one of the most important reasons for our design choice.
Some cents regarding the default operator resource: - It might be good for the scenario of DataStream jobs. ** For light-weight operators, the accumulative configuration error will not be significant. Then, the resource of a task used is proportional to the number of operators it contains. ** For heavy operators like join and window or operators using the external resources, user will turn to the fine-grained resource configuration. - It can increase the stability for the standalone cluster where task executors registered are heterogeneous(with different default slot resources). - It might not be good for SQL users. The operators that SQL will be transferred to is a black box to the user. We also do not guarantee the cross-version of consistency of the transformation so far. I think it can be treated as a follow-up work when the fine-grained resource management is end-to-end ready. Best, Yangze Guo On Wed, Jan 20, 2021 at 11:16 AM Xintong Song <tonysong...@gmail.com> wrote: > > Thanks for the feedback, Till. > > ## I feel that what you proposed (operator-based + default value) might be > subsumed by the SSG-based approach. > Thinking of op_1 -> op_2, there are the following 4 cases, categorized by > whether the resource requirements are known to the users. > > 1. *Both known.* As previously mentioned, there's no reason to put > multiple operators whose individual resource requirements are already known > into the same group in fine-grained resource management. And if op_1 and > op_2 are in different groups, there should be no problem switching data > exchange mode from pipelined to blocking. This is equivalent to specifying > operator resource requirements in your proposal. > 2. *op_1 known, op_2 unknown.* Similar to 1), except that op_2 is in a > SSG whose resource is not specified thus would have the default slot > resource. This is equivalent to having default operator resources in your > proposal. > 3. *Both unknown*. The user can either set op_1 and op_2 to the same SSG > or separate SSGs. > - If op_1 and op_2 are in the same SSG, it will be equivalent to the > coarse-grained resource management, where op_1 and op_2 share a default > size slot no matter which data exchange mode is used. > - If op_1 and op_2 are in different SSGs, then each of them will use > a default size slot. This is equivalent to setting them with default > operator resources in your proposal. > 4. *Total (pipeline) or max (blocking) of op_1 and op_2 is known.* > - It is possible that the user learns the total / max resource > requirement from executing and monitoring the job, while not > being aware of > individual operator requirements. > - I believe this is the case your proposal does not cover. And TBH, > this is probably how most users learn the resource requirements, > according > to my experiences. > - In this case, the user might need to specify different resources if > he wants to switch the execution mode, which should not be worse than > not > being able to use fine-grained resource management. > > > ## An additional idea inspired by your proposal. > We may provide multiple options for deciding resources for SSGs whose > requirement is not specified, if needed. > > - Default slot resource (current design) > - Default operator resource times number of operators (equivalent to > your proposal) > > > ## Exposing internal runtime strategies > Theoretically, yes. Tying to the SSGs, the resource requirements might be > affected if how SSGs are internally handled changes in future. Practically, > I do not concretely see at the moment what kind of changes we may want in > future that might conflict with this FLIP proposal, as the question of > switching data exchange mode answered above. I'd suggest to not give up the > user friendliness we may gain now for the future problems that may or may > not exist. > > Moreover, the SSG-based approach has the flexibility to achieve the > equivalent behavior as the operator-based approach, if we set each operator > (or task) to a separate SSG. We can even provide a shortcut option to > automatically do that for users, if needed. > > > Thank you~ > > Xintong Song > > > > On Tue, Jan 19, 2021 at 11:48 PM Till Rohrmann <trohrm...@apache.org> wrote: > > > Thanks for the responses Xintong and Stephan, > > > > I agree that being able to define the resource requirements for a group of > > operators is more user friendly. However, my concern is that we are > > exposing thereby internal runtime strategies which might limit our > > flexibility to execute a given job. Moreover, the semantics of configuring > > resource requirements for SSGs could break if switching from streaming to > > batch execution. If one defines the resource requirements for op_1 -> op_2 > > which run in pipelined mode when using the streaming execution, then how do > > we interpret these requirements when op_1 -> op_2 are executed with a > > blocking data exchange in batch execution mode? Consequently, I am still > > leaning towards Stephan's proposal to set the resource requirements per > > operator. > > > > Maybe the following proposal makes the configuration easier: If the user > > wants to use fine-grained resource requirements, then she needs to specify > > the default size which is used for operators which have no explicit > > resource annotation. If this holds true, then every operator would have a > > resource requirement and the system can try to execute the operators in the > > best possible manner w/o being constrained by how the user set the SSG > > requirements. > > > > Cheers, > > Till > > > > On Tue, Jan 19, 2021 at 9:09 AM Xintong Song <tonysong...@gmail.com> > > wrote: > > > > > Thanks for the feedback, Stephan. > > > > > > Actually, your proposal has also come to my mind at some point. And I > > have > > > some concerns about it. > > > > > > > > > 1. It does not give users the same control as the SSG-based approach. > > > > > > > > > While both approaches do not require specifying for each operator, > > > SSG-based approach supports the semantic that "some operators together > > use > > > this much resource" while the operator-based approach doesn't. > > > > > > > > > Think of a long pipeline with m operators (o_1, o_2, ..., o_m), and at > > some > > > point there's an agg o_n (1 < n < m) which significantly reduces the data > > > amount. One can separate the pipeline into 2 groups SSG_1 (o_1, ..., o_n) > > > and SSG_2 (o_n+1, ... o_m), so that configuring much higher parallelisms > > > for operators in SSG_1 than for operators in SSG_2 won't lead to too much > > > wasting of resources. If the two SSGs end up needing different resources, > > > with the SSG-based approach one can directly specify resources for the > > two > > > groups. However, with the operator-based approach, the user will have to > > > specify resources for each operator in one of the two groups, and tune > > the > > > default slot resource via configurations to fit the other group. > > > > > > > > > 2. It increases the chance of breaking operator chains. > > > > > > > > > Setting chainnable operators into different slot sharing groups will > > > prevent them from being chained. In the current implementation, > > downstream > > > operators, if SSG not explicitly specified, will be set to the same group > > > as the chainable upstream operators (unless multiple upstream operators > > in > > > different groups), to reduce the chance of breaking chains. > > > > > > > > > Thinking of chainable operators o_1 -> o_2 -> o_3 -> o_3, deciding SSGs > > > based on whether resource is specified we will easily get groups like > > (o_1, > > > o_3) & (o_2, o_4), where none of the operators can be chained. This is > > also > > > possible for the SSG-based approach, but I believe the chance is much > > > smaller because there's no strong reason for users to specify the groups > > > with alternate operators like that. We are more likely to get groups like > > > (o_1, o_2) & (o_3, o_4), where the chain breaks only between o_2 and o_3. > > > > > > > > > 3. It complicates the system by having two different mechanisms for > > sharing > > > managed memory in a slot. > > > > > > > > > - In FLIP-141, we introduced the intra-slot managed memory sharing > > > mechanism, where managed memory is first distributed according to the > > > consumer type, then further distributed across operators of that consumer > > > type. > > > > > > - With the operator-based approach, managed memory size specified for an > > > operator should account for all the consumer types of that operator. That > > > means the managed memory is first distributed across operators, then > > > distributed to different consumer types of each operator. > > > > > > > > > Unfortunately, the different order of the two calculation steps can lead > > to > > > different results. To be specific, the semantic of the configuration > > option > > > `consumer-weights` changed (within a slot vs. within an operator). > > > > > > > > > > > > To sum up things: > > > > > > While (3) might be a bit more implementation related, I think (1) and (2) > > > somehow suggest that, the price for the proposed approach to avoid > > > specifying resource for every operator is that it's not as independent > > from > > > operator chaining and slot sharing as the operator-based approach > > discussed > > > in the FLIP. > > > > > > > > > Thank you~ > > > > > > Xintong Song > > > > > > > > > > > > On Tue, Jan 19, 2021 at 4:29 AM Stephan Ewen <se...@apache.org> wrote: > > > > > > > Thanks a lot, Yangze and Xintong for this FLIP. > > > > > > > > I want to say, first of all, that this is super well written. And the > > > > points that the FLIP makes about how to expose the configuration to > > users > > > > is exactly the right thing to figure out first. > > > > So good job here! > > > > > > > > About how to let users specify the resource profiles. If I can sum the > > > FLIP > > > > and previous discussion up in my own words, the problem is the > > following: > > > > > > > > Operator-level specification is the simplest and cleanest approach, > > > because > > > > > it avoids mixing operator configuration (resource) and scheduling. No > > > > > matter what other parameters change (chaining, slot sharing, > > switching > > > > > pipelined and blocking shuffles), the resource profiles stay the > > same. > > > > > But it would require that a user specifies resources on all > > operators, > > > > > which makes it hard to use. That's why the FLIP suggests going with > > > > > specifying resources on a Sharing-Group. > > > > > > > > > > > > I think both thoughts are important, so can we find a solution where > > the > > > > Resource Profiles are specified on an Operator, but we still avoid that > > > we > > > > need to specify a resource profile on every operator? > > > > > > > > What do you think about something like the following: > > > > - Resource Profiles are specified on an operator level. > > > > - Not all operators need profiles > > > > - All Operators without a Resource Profile ended up in the default > > slot > > > > sharing group with a default profile (will get a default slot). > > > > - All Operators with a Resource Profile will go into another slot > > > sharing > > > > group (the resource-specified-group). > > > > - Users can define different slot sharing groups for operators like > > > they > > > > do now, with the exception that you cannot mix operators that have a > > > > resource profile and operators that have no resource profile. > > > > - The default case where no operator has a resource profile is just a > > > > special case of this model > > > > - The chaining logic sums up the profiles per operator, like it does > > > now, > > > > and the scheduler sums up the profiles of the tasks that it schedules > > > > together. > > > > > > > > > > > > There is another question about reactive scaling raised in the FLIP. I > > > need > > > > to think a bit about that. That is indeed a bit more tricky once we > > have > > > > slots of different sizes. > > > > It is not clear then which of the different slot requests the > > > > ResourceManager should fulfill when new resources (TMs) show up, or how > > > the > > > > JobManager redistributes the slots resources when resources (TMs) > > > disappear > > > > This question is pretty orthogonal, though, to the "how to specify the > > > > resources". > > > > > > > > > > > > Best, > > > > Stephan > > > > > > > > On Fri, Jan 8, 2021 at 5:14 AM Xintong Song <tonysong...@gmail.com> > > > wrote: > > > > > > > > > Thanks for drafting the FLIP and driving the discussion, Yangze. > > > > > And Thanks for the feedback, Till and Chesnay. > > > > > > > > > > @Till, > > > > > > > > > > I agree that specifying requirements for SSGs means that SSGs need to > > > be > > > > > supported in fine-grained resource management, otherwise each > > operator > > > > > might use as many resources as the whole group. However, I cannot > > think > > > > of > > > > > a strong reason for not supporting SSGs in fine-grained resource > > > > > management. > > > > > > > > > > > > > > > > Interestingly, if all operators have their resources properly > > > > specified, > > > > > > then slot sharing is no longer needed because Flink could slice off > > > the > > > > > > appropriately sized slots for every Task individually. > > > > > > > > > > > > > > > > So for example, if we have a job consisting of two operator op_1 and > > > op_2 > > > > > > where each op needs 100 MB of memory, we would then say that the > > slot > > > > > > sharing group needs 200 MB of memory to run. If we have a cluster > > > with > > > > 2 > > > > > > TMs with one slot of 100 MB each, then the system cannot run this > > > job. > > > > If > > > > > > the resources were specified on an operator level, then the system > > > > could > > > > > > still make the decision to deploy op_1 to TM_1 and op_2 to TM_2. > > > > > > > > > > > > > > > Couldn't agree more that if all operators' requirements are properly > > > > > specified, slot sharing should be no longer needed. I think this > > > exactly > > > > > disproves the example. If we already know op_1 and op_2 each needs > > 100 > > > MB > > > > > of memory, why would we put them in the same group? If they are in > > > > separate > > > > > groups, with the proposed approach the system can freely deploy them > > to > > > > > either a 200 MB TM or two 100 MB TMs. > > > > > > > > > > Moreover, the precondition for not needing slot sharing is having > > > > resource > > > > > requirements properly specified for all operators. This is not always > > > > > possible, and usually requires tremendous efforts. One of the > > benefits > > > > for > > > > > SSG-based requirements is that it allows the user to freely decide > > the > > > > > granularity, thus efforts they want to pay. I would consider SSG in > > > > > fine-grained resource management as a group of operators that the > > user > > > > > would like to specify the total resource for. There can be only one > > > group > > > > > in the job, 2~3 groups dividing the job into a few major parts, or as > > > > many > > > > > groups as the number of tasks/operators, depending on how > > fine-grained > > > > the > > > > > user is able to specify the resources. > > > > > > > > > > Having to support SSGs might be a constraint. But given that all the > > > > > current scheduler implementations already support SSGs, I tend to > > think > > > > > that as an acceptable price for the above discussed usability and > > > > > flexibility. > > > > > > > > > > @Chesnay > > > > > > > > > > Will declaring them on slot sharing groups not also waste resources > > if > > > > the > > > > > > parallelism of operators within that group are different? > > > > > > > > > > > Yes. It's a trade-off between usability and resource utilization. To > > > > avoid > > > > > such wasting, the user can define more groups, so that each group > > > > contains > > > > > less operators and the chance of having operators with different > > > > > parallelism will be reduced. The price is to have more resource > > > > > requirements to specify. > > > > > > > > > > It also seems like quite a hassle for users having to recalculate the > > > > > > resource requirements if they change the slot sharing. > > > > > > I'd think that it's not really workable for users that create a set > > > of > > > > > > re-usable operators which are mixed and matched in their > > > applications; > > > > > > managing the resources requirements in such a setting would be a > > > > > > nightmare, and in the end would require operator-level requirements > > > any > > > > > > way. > > > > > > In that sense, I'm not even sure whether it really increases > > > usability. > > > > > > > > > > > > > > > > - As mentioned in my reply to Till's comment, there's no reason to > > > put > > > > > multiple operators whose individual resource requirements are > > > already > > > > > known > > > > > into the same group in fine-grained resource management. > > > > > - Even an operator implementation is reused for multiple > > > applications, > > > > > it does not guarantee the same resource requirements. During our > > > years > > > > > of > > > > > practices in Alibaba, with per-operator requirements specified for > > > > > Blink's > > > > > fine-grained resource management, very few users (including our > > > > > specialists > > > > > who are dedicated to supporting Blink users) are as experienced as > > > to > > > > > accurately predict/estimate the operator resource requirements. > > Most > > > > > people > > > > > rely on the execution-time metrics (throughput, delay, cpu load, > > > > memory > > > > > usage, GC pressure, etc.) to improve the specification. > > > > > > > > > > To sum up: > > > > > If the user is capable of providing proper resource requirements for > > > > every > > > > > operator, that's definitely a good thing and we would not need to > > rely > > > on > > > > > the SSGs. However, that shouldn't be a *must* for the fine-grained > > > > resource > > > > > management to work. For those users who are capable and do not like > > > > having > > > > > to set each operator to a separate SSG, I would be ok to have both > > > > > SSG-based and operator-based runtime interfaces and to only fallback > > to > > > > the > > > > > SSG requirements when the operator requirements are not specified. > > > > However, > > > > > as the first step, I think we should prioritise the use cases where > > > users > > > > > are not that experienced. > > > > > > > > > > Thank you~ > > > > > > > > > > Xintong Song > > > > > > > > > > On Thu, Jan 7, 2021 at 9:55 PM Chesnay Schepler <ches...@apache.org> > > > > > wrote: > > > > > > > > > > > Will declaring them on slot sharing groups not also waste resources > > > if > > > > > > the parallelism of operators within that group are different? > > > > > > > > > > > > It also seems like quite a hassle for users having to recalculate > > the > > > > > > resource requirements if they change the slot sharing. > > > > > > I'd think that it's not really workable for users that create a set > > > of > > > > > > re-usable operators which are mixed and matched in their > > > applications; > > > > > > managing the resources requirements in such a setting would be a > > > > > > nightmare, and in the end would require operator-level requirements > > > any > > > > > > way. > > > > > > In that sense, I'm not even sure whether it really increases > > > usability. > > > > > > > > > > > > My main worry is that it if we wire the runtime to work on SSGs > > it's > > > > > > gonna be difficult to implement more fine-grained approaches, which > > > > > > would not be the case if, for the runtime, they are always defined > > on > > > > an > > > > > > operator-level. > > > > > > > > > > > > On 1/7/2021 2:42 PM, Till Rohrmann wrote: > > > > > > > Thanks for drafting this FLIP and starting this discussion > > Yangze. > > > > > > > > > > > > > > I like that defining resource requirements on a slot sharing > > group > > > > > makes > > > > > > > the overall setup easier and improves usability of resource > > > > > requirements. > > > > > > > > > > > > > > What I do not like about it is that it changes slot sharing > > groups > > > > from > > > > > > > being a scheduling hint to something which needs to be supported > > in > > > > > order > > > > > > > to support fine grained resource requirements. So far, the idea > > of > > > > slot > > > > > > > sharing groups was that it tells the system that a set of > > operators > > > > can > > > > > > be > > > > > > > deployed in the same slot. But the system still had the freedom > > to > > > > say > > > > > > that > > > > > > > it would rather place these tasks in different slots if it > > wanted. > > > If > > > > > we > > > > > > > now specify resource requirements on a per slot sharing group, > > then > > > > the > > > > > > > only option for a scheduler which does not support slot sharing > > > > groups > > > > > is > > > > > > > to say that every operator in this slot sharing group needs a > > slot > > > > with > > > > > > the > > > > > > > same resources as the whole group. > > > > > > > > > > > > > > So for example, if we have a job consisting of two operator op_1 > > > and > > > > > op_2 > > > > > > > where each op needs 100 MB of memory, we would then say that the > > > slot > > > > > > > sharing group needs 200 MB of memory to run. If we have a cluster > > > > with > > > > > 2 > > > > > > > TMs with one slot of 100 MB each, then the system cannot run this > > > > job. > > > > > If > > > > > > > the resources were specified on an operator level, then the > > system > > > > > could > > > > > > > still make the decision to deploy op_1 to TM_1 and op_2 to TM_2. > > > > > > > > > > > > > > Originally, one of the primary goals of slot sharing groups was > > to > > > > make > > > > > > it > > > > > > > easier for the user to reason about how many slots a job needs > > > > > > independent > > > > > > > of the actual number of operators in the job. Interestingly, if > > all > > > > > > > operators have their resources properly specified, then slot > > > sharing > > > > is > > > > > > no > > > > > > > longer needed because Flink could slice off the appropriately > > sized > > > > > slots > > > > > > > for every Task individually. What matters is whether the whole > > > > cluster > > > > > > has > > > > > > > enough resources to run all tasks or not. > > > > > > > > > > > > > > Cheers, > > > > > > > Till > > > > > > > > > > > > > > On Thu, Jan 7, 2021 at 4:08 AM Yangze Guo <karma...@gmail.com> > > > > wrote: > > > > > > > > > > > > > >> Hi, there, > > > > > > >> > > > > > > >> We would like to start a discussion thread on "FLIP-156: Runtime > > > > > > >> Interfaces for Fine-Grained Resource Requirements"[1], where we > > > > > > >> propose Slot Sharing Group (SSG) based runtime interfaces for > > > > > > >> specifying fine-grained resource requirements. > > > > > > >> > > > > > > >> In this FLIP: > > > > > > >> - Expound the user story of fine-grained resource management. > > > > > > >> - Propose runtime interfaces for specifying SSG-based resource > > > > > > >> requirements. > > > > > > >> - Discuss the pros and cons of the three potential granularities > > > for > > > > > > >> specifying the resource requirements (op, task and slot sharing > > > > group) > > > > > > >> and explain why we choose the slot sharing group. > > > > > > >> > > > > > > >> Please find more details in the FLIP wiki document [1]. Looking > > > > > > >> forward to your feedback. > > > > > > >> > > > > > > >> [1] > > > > > > >> > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-156%3A+Runtime+Interfaces+for+Fine-Grained+Resource+Requirements > > > > > > >> > > > > > > >> Best, > > > > > > >> Yangze Guo > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > >
