## configuration option I see your point that autoscaling mode might be more suitable for session clusters. It doesn't change that `execution-mode` could be a job-level configuration. But I'm good with keeping it cluster-level and marking it experimental at the moment, so we can change it later if needed for the autoscaling mode.
## active resource managers I'm a bit confused about the boundary between reactive mode and autoscaling mode. - Reactive mode requests an infinite amount of resources, and executes at the largest parallelism that is possible with the available resources. - Autoscaling mode dynamically adjusts resource demand, and executes at a parallelism that is either demanded or as large as possible if the demanded parallelism cannot be reached. - What about something in between? A job is not capable of dynamically adjusting the resource demand and requests a fixed finite amount of resources, and still wants to be executed with as large parallelisms as possible if the demanded parallelism cannot be reached? It's quite common that a job may temporarily not get as much resources as desired, due to running of other higher priority jobs in the Kubernetes/Yarn/Mesos cluster. In such cases, currently either the user needs to configure the job with a different parallelism, or the job cannot be executed. It would be helpful if the job can execute with a lower parallelism, and automatically scales up to the original desired parallelism when more resources become available. For Yarn, there's comprehensive queue based resource quota management, where how many resources each job gets are closely related to other jobs' resource requirements. For Kubernetes, while the default kube-scheduler does not have such mature multi-tenant support, there are other projects (e.g., Apache YuniKorn [1]) that can bring the similar scheduling capability to Kubernetes Thank you~ Xintong Song [1] https://yunikorn.apache.org/ On Mon, Jan 25, 2021 at 4:48 PM Robert Metzger <rmetz...@apache.org> wrote: > Thank you very much for the comments so far. > > @Steven: > > No fixed parallelism for any of the operators > > > > Regarding this limitation, can the scheduler only adjust the default > > parallelism? if some operators set parallelism explicitly (like always > 1), > > just leave them unchanged. > > > We will respect the configured maxParallelism for that purpose. If you have > an operator that is not intended to run in parallel, you can set > maxParalellism=1. > > @Xintong: > > the cluster configuration option will limit us from having jobs running > > with different execution modes in the same session cluster. > > > I'm not sure if it makes sense to support reactive mode in a session > cluster ever. For an autoscaling mode, it probably makes sense (as we can > just combine the resource requests from all running jobs, and distribute > the available resources proportional to the requested resources). > > I will state more clearly in the FLIP that the configuration options should > be marked as experimental. > > Active resource managers > > [...] > > If this is the only concern, I'd like to bring the configuration option > > `slotmanager.number-of-slots.max` to your attention. > > > I understand and agree that it would be really nice to support active > resource managers with the new scheduler right away. In my opinion, > reactive mode will never be really supported by active resource managers, > as this is a contradiction with the idea of reactive mode: It is explicitly > designed to allow controlling the cluster from the outside (similar to > Kafka streams, where you add and remove capacity for scaling). Integration > with active resource managers should be added in a autoscaling mode, based > on the declarative scheduler. > I've considered the slotmanager.number-of-slots.max option as well, but it > basically means that your cluster will always immediately scale up > to slotmanager.number-of-slots.max and stick to that value, even if those > resources are not needed. > On YARN, it would be pretty difficult or even impossible to control the > scale of such a Flink deployment from the outside (using a queue with the > capacity scheduler won't work, as changes to queues require restarts) > On K8s, one would have to build a custom tool that finds the deployment > created by Flink and adjusts it. Then, it's probably easier to just create > a standalone deployment on K8s. > > @Yang: > > It will be better to make the 10 seconds to be configurable. > > > I agree that it is pretty bold to have such an important configuration > parameter hardcoded. We proposed it like this to keep the first > implementation as simple as possible. > But if we see that basically everybody is asking for this, or if we have > time left at the end of the release cycle, we'll make it configurable. > > > but also the ScalingPolicy is not exposed to the users now > > > Exposing the ScalingPolicy to the user is very high on our priority list, > but we want to keep the first version as simple as possible, to be able to > deliver the overall feature in time, and to collect some initial user > feedback before coming up with an interface we want to expose. > > > > > > > > On Mon, Jan 25, 2021 at 8:37 AM Yang Wang <danrtsey...@gmail.com> wrote: > > > Thanks Robert for creating this FLIP and starting the discussion. > > > > This is a great start point to make Flink work with auto scaling service. > > The reactive mode > > is very useful in containerized environment(e.g. docker, Kubernetes). For > > example, combined > > with Kubernetes "Horizontal Pod Autoscaler"[1], the TaskManagers could be > > started/released > > dynamically based on the system metrics(e.g. cpu, memory) and custom > > metrics(e.g. delay, latency). > > > > > > > Once the job has started running, and a TaskManager is lost, it will > wait > > > for 10 seconds for the > > > > TaskManager to re-appear. > > > > It will be better to make the 10 seconds to be configurable. According to > > our production experience > > on Kubernetes, 10 seconds is not enough for a pod to be relaunched. Maybe > > this is also a specific > > case whether the resource is stable or not. > > > > > Active ResourceManager > > > > IIUC, the reason why reactive mode could not work with active resource > > manager is not only > > about requesting infinite amount of resources, but also the ScalingPolicy > > is not exposed to the > > users now. ScalingPolicy could be the bridge between reactive mode and > > active resource manager. > > User could have their own auto scaling service, which monitor the Flink > > metrics and then update > > the ScalingPolicy(e.g. parallelism 10 -> 20). Then the active resource > > manager could allocate these > > TaskManagers. > > But it is out the scope of this FLIP, I really expect this could be done > in > > the future. And it will be another > > great step to make Flink auto scalable. > > > > > > > > [1]. > > > https://kubernetes.io/docs/tasks/run-application/horizontal-pod-autoscale/ > > > > > > Best, > > Yang > > > > > > Xintong Song <tonysong...@gmail.com> 于2021年1月25日周一 上午10:29写道： > > > > > Thanks for preparing the FLIP and starting the discussion, Robert. > > > > > > ## Cluster vs. Job configuration > > > As I have commented on the FLIP-160 discussion thread [1], I'm a bit > > unsure > > > about activating the reactive execution mode via a cluster level > > > configuration option. I'm aware that in the first step this feature > does > > > not support session clusters. However, I think that does not mean it > > won't > > > be supported in future. In that case, the cluster configuration option > > will > > > limit us from having jobs running with different execution modes in the > > > same session cluster. > > > > > > ## Active resource managers > > > According to the FLIP, this feature explicitly does not support active > > > resource managers. IIUC, this is because when in this feature the job > > > requests an infinite amount of resources, which would flood Kubernetes > / > > > Yarn / Mesos with unreasonably large number of resource requests. If > this > > > is the only concern, I'd like to bring the configuration option > > > `slotmanager.number-of-slots.max` to your attention. This feature > allows > > > putting an upper limit to the total number of slots the Flink cluster > > uses, > > > preventing active resource managers from allocating too many resources > > from > > > Kubernetes / Yarn / Mesos. Unless there are other concerns that I > > > overlooked, I think it would be nicer for the reactive mode to also > > support > > > active resource managers, with the additional requirement to explicitly > > > configure the max slots. > > > > > > Thank you~ > > > > > > Xintong Song > > > > > > > > > [1] > > > > > > > > > http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-FLIP-160-Declarative-scheduler-td48165.html > > > > > > On Sat, Jan 23, 2021 at 5:59 AM Steven Wu <stevenz...@gmail.com> > wrote: > > > > > > > Thanks a lot for the proposal, Robert and Till. > > > > > > > > > No fixed parallelism for any of the operators > > > > > > > > Regarding this limitation, can the scheduler only adjust the default > > > > parallelism? if some operators set parallelism explicitly (like > always > > > 1), > > > > just leave them unchanged. > > > > > > > > > > > > On Fri, Jan 22, 2021 at 8:42 AM Robert Metzger <rmetz...@apache.org> > > > > wrote: > > > > > > > > > Hi all, > > > > > > > > > > Till started a discussion about FLIP-160: Declarative scheduler [1] > > > > earlier > > > > > today, the first major feature based on that effort will be > FLIP-159: > > > > > Reactive Mode. It allows users to operate Flink in a way that it > > > > reactively > > > > > scales the job up or down depending on the provided resources: > adding > > > > > TaskManagers will scale the job up, removing them will scale it > down > > > > again. > > > > > > > > > > Here's the link to the Wiki: > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-159%3A+Reactive+Mode > > > > > > > > > > We are very excited to hear your feedback about the proposal! > > > > > > > > > > Best, > > > > > Robert > > > > > > > > > > [1] > > > > > > > > > > > > > > > > > > > > https://lists.apache.org/thread.html/r604a01f739639e2a5f093fbe7894c172125530332747ecf6990a6ce4%40%3Cdev.flink.apache.org%3E > > > > > > > > > > > > > > >
