You are viewing a plain text version of this content. The canonical link for it is here.

Posted to dev@flink.apache.org by Maximilian Michels <mx...@apache.org> on 2023/01/23 17:07:43 UTC

Reworking the Rescale API

Hi,

The current rescale API appears to be a work in progress. A couple years
ago, we disabled access to the API [1].

I'm looking into this problem as part of working on autoscaling [2] where
we currently require a full restart of the job to apply the parallelism
overrides. This adds additional delay and comes with the caveat that we
don't know whether sufficient resources are available prior to executing
the scaling decision. We obviously do not want to get stuck due to a lack
of resources. So a rescale API would have to ensure enough resources are
available prior to restarting the job.

I've created an issue here:
https://issues.apache.org/jira/browse/FLINK-30773

Any comments or interest in working on this?

-Max

[1] https://issues.apache.org/jira/browse/FLINK-12312
[2]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling

Re: Reworking the Rescale API

Posted by Maximilian Michels <mx...@apache.org>.

>> I fully agree that in-place scaling is a much harder problem which is out of the scope for now. My primary concern here is to be able to rescale with upfront reservation of resources before restarting the job, so the job doesn't get stuck in case of resource constraints.
> Not sure I follow. The AS only rescales when it has already acquired the slots that it needs.

I'm saying that the primary objective of this thread is to figure out
upfront reservation of resources as part of a new Rescale API. The
adaptive scheduler is a (very reasonable) means to an end to fulfil
this property. If we were to go with another solution because the
adaptive scheduler does not prove to be production ready, then we
would still have to make that property holds. I'm going to experiment
a bit with the adaptive scheduler to see if there are any other
limitations.

As for the slot sharing groups with different maximum parallelism, I
see what the issue is here:
https://github.com/apache/flink/blob/2ae5df278958073fee63b2bf824a53a28a21701b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java#L97
Should be fixable. I've filed a JIRA here:
https://issues.apache.org/jira/browse/FLINK-30931

-Max

On Fri, Feb 3, 2023 at 10:13 AM Chesnay Schepler <ch...@apache.org> wrote:
>
> > My primary concern here is to be able to rescale with upfront reservation of resources before restarting the job, so the job doesn't get stuck in case of resource constraints.
>
> Not sure I follow. The AS only rescales when it has already acquired the slots that it needs.
>
>  > This is a blocker from my side. Why do we have that restriction?
>
> We just didn't bother fixing it initially. It should be easy to fix.
>
> On 02/02/2023 18:29, Maximilian Michels wrote:
> > I fully agree that in-place scaling is a much harder problem which is
> > out of the scope for now. My primary concern here is to be able to
> > rescale with upfront reservation of resources before restarting the
> > job, so the job doesn't get stuck in case of resource constraints.
> >
> >> Unused slots: If the max parallelism for slot sharing groups is not equal, slots offered to Adaptive Scheduler might be unused.
> > From: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/#limitations-1
> >
> > This is a blocker from my side. Why do we have that restriction?
> >
> > On Thu, Feb 2, 2023 at 5:03 PM Chesnay Schepler <ch...@apache.org> wrote:
> >>   > If I understand correctly, the adaptive scheduler currently does a
> >> full job restart. Is there any work planned to enable in-place rescaling
> >> with the adaptive scheduler?
> >>
> >> Nothing concrete. Sure, it's on a wishlist, but it'd require significant
> >> changes to how the runtime works.
> >> Rescaling stateful operators requires keygroups to be redistributed,
> >> you'd need to be able to change task edges dynamically, roll-back to a
> >> checkpoint without restarting tasks, ...
> >>
> >> It's less of a scheduler thing actually.
> >>
> >> An earlier step to that would be to allow recovery from an error without
> >> restarting all tasks, which would benefit all schedulers.
> >> But again bit of a moonshot.
> >>
> >>   > How well has the adaptive scheduler been tested in production? If we
> >> are intending to use it for rescale operations, I'm a bit concerned
> >> those jobs might show different behavior due to the scheduling than jobs
> >> started with the default scheduler.
> >>
> >> I don't think we got a lot of feedback so far.
> >> Outside of the limitations listed on the elastic scaling page (which I
> >> believe we'll address in due time) I'm not aware of any problems.
> >> We haven't run into any issues internally.
> >>
> >> On 02/02/2023 12:44, Maximilian Michels wrote:
> >>> +1 on improving the scheduler docs.
> >>>
> >>>> They never shared a base class since day 1. Are you maybe mixing up the AdaptiveScheduler and AdaptiveBatchScheduler?
> >>> @Chesnay: Indeed, I had mixed this up. DefaultScheduler and
> >>> AdaptiveScheduler only share the SchedulerNG interface while the
> >>> DefaultScheduler and the AdaptiveBatchScheduler share a subset of the
> >>> code. Too many schedulers :)
> >>>
> >>> Thanks for clarifying the current and the intended feature set of the
> >>> adaptive scheduler!
> >>>
> >>> How well has the adaptive scheduler been tested in production? If we
> >>> are intending to use it for rescale operations, I'm a bit concerned
> >>> those jobs might show different behavior due to the scheduling than
> >>> jobs started with the default scheduler.
> >>>
> >>> If I understand correctly, the adaptive scheduler currently does a
> >>> full job restart. Is there any work planned to enable in-place
> >>> rescaling with the adaptive scheduler?
> >>>
> >>>> @max:
> >>>>     - when user repartition, we still need to restart the job, can we try to
> >>>>     do this part of the work internally instead of externally, as
> >>>>     *@konstantin* said only trigger rescaling when the checkpoint or
> >>>>     retain-checkpoint is completed operations to minimize reprocessing
> >>> @ConradJam: I'm not sure I understand your question. Do you mean when
> >>> the partition strategy changes between operators? That shouldn't be
> >>> the case for Rescale (except maybe converting ForwardPartitioner to
> >>> RescalePartitioner). A more advanced rescale API could allow user
> >>> control over this but for now I think it would only support adjusting
> >>> parallelism of vertices.
> >>>
> >>> -Max
> >>>
> >>> On Thu, Feb 2, 2023 at 6:44 AM weijie guo <gu...@gmail.com> wrote:
> >>>> Hi David,
> >>>>
> >>>> Sorry I'm late to join discuss.
> >>>>
> >>>> +1 for having a more structure doc about scheduler ecosystem and I can help to fill in the details about batch part.
> >>>>
> >>>> Best regards,
> >>>>
> >>>> Weijie
> >>>>
> >>>>
> >>>>
> >>>> David Morávek <dm...@apache.org> 于2023年2月1日周三 22:38写道：
> >>>>> It makes sense to give the whole "scheduler ecosystem," not just the
> >>>>> adaptive scheduler, a little bit more structure in the docs. We already
> >>>>> have 4 different schedulers (Default, Adaptive, AdaptiveBatch,
> >>>>> AdaptiveBatchSpeculative), and it becomes quite confusing since the details
> >>>>> are scattered around the docs. Maybe having a "Job Schedulers" subpage, the
> >>>>> same way as we have for "Resource Providers" could do the trick.
> >>>>>
> >>>>> I should be able to fill in the details about the streaming ones, but I
> >>>>> will probably need some help with the batch ones.
> >>>>>
> >>>>> As for the first FLIP, it's already prepared and we should be able to
> >>>>> publish it until Friday.
> >>>>>
> >>>>> Best,
> >>>>> D.
> >>>>>
> >>>>>
> >>>>> On Wed, Feb 1, 2023 at 9:56 AM Gyula Fóra <gy...@gmail.com> wrote:
> >>>>>
> >>>>>> Chesnay, David:
> >>>>>>
> >>>>>> Thank you guys for the extra information. We were clearly missing some
> >>>>>> context here around the scheduler related efforts and the currently
> >>>>>> available feature set.
> >>>>>>
> >>>>>> As for the concrete suggestions regarding the docs.
> >>>>>>
> >>>>>> 1. If the adaptive scheduler provides a significantly different feature set
> >>>>>> from the default scheduler we could have its own smaller doc page detailing
> >>>>>> the differences and why people should switch to it for streaming. This will
> >>>>>> also help us when we are making the transition and change the default
> >>>>>> behaviour.
> >>>>>> 2. We could still have an elastic scaling page that links to the adaptive
> >>>>>> scheduler (and vice versa) that focuses on elastic scaling + the Kubernetes
> >>>>>> operator autoscaler for a complete picture on elastic scaling options +
> >>>>>> detailing the limitations of the different approaches.
> >>>>>>
> >>>>>> This way the Adaptive Scheduler docs will be decoupled from elastic scaling
> >>>>>> and will result in a better understanding for the users (it sure would have
> >>>>>> helped us here, and we are on the more advanced user side :))
> >>>>>>
> >>>>>> What do you think?
> >>>>>> Gyula
> >>>>>>
> >>>>>> On Sat, Jan 28, 2023 at 4:20 AM ConradJam <ja...@gmail.com> wrote:
> >>>>>>
> >>>>>>> Sorry I'm late to join discuss, I've gleaned a lot of useful information
> >>>>>>> from you guys
> >>>>>>>
> >>>>>>> *@max*
> >>>>>>>
> >>>>>>>      - when user repartition, we still need to restart the job, can we try
> >>>>>> to
> >>>>>>>      do this part of the work internally instead of externally, as
> >>>>>>>      *@konstantin* said only trigger rescaling when the checkpoint or
> >>>>>>>      retain-checkpoint is completed operations to minimize reprocessing
> >>>>>>>
> >>>>>>> *@konstantin*
> >>>>>>>
> >>>>>>>      - I think you mentioned that 2 FLIPs are being drafted which I
> >>>>>> consider
> >>>>>>>      to be the condition to achieve the *@max* goal, I would love to join
> >>>>>>>      this discussion and contribute it. I've tried a native implementation
> >>>>>> of
> >>>>>>>      this part myself, if I can help the community that's the best I can do
> >>>>>>>
> >>>>>>> *@chesnay*
> >>>>>>>
> >>>>>>>      - The docs section is confusion/misconceptions confusing like *@gyula
> >>>>>>> *say,
> >>>>>>>      I'll see if I can fix it
> >>>>>>>
> >>>>>>>
> >>>>>>> *About Rescale Api*
> >>>>>>>
> >>>>>>>     Some limitations and differences between *default* and *reactive mode*
> >>>>>>> were
> >>>>>>> discussed earlier, and *@chesnay* explained some of their limitations and
> >>>>>>> behaviors, essentially they are two different things. I agree that when
> >>>>>>> reactive mode is ready, it should be used as the *reactive mode* for the
> >>>>>>> default *stream processing* job.
> >>>>>>>     As for the *[1] **Rescale API*, as we know now it seems to be
> >>>>>> unusable, I
> >>>>>>> believe the goal of this api is to be able to do fast reparallelism. I
> >>>>>>> would like to wait until the discussion is over and the 2 draft FILPs
> >>>>>>> mentioned earlier are completed. It is not too late to make another
> >>>>>>> decision on whether to modify the *[2] **Rescale Rest API *to support for
> >>>>>>> parallelism modification of job vertices
> >>>>>>>
> >>>>>>>
> >>>>>>>      1.
> >>>>>>> *
> >>>>>>>
> >>>>>> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
> >>>>>>>      <
> >>>>>>>
> >>>>>> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
> >>>>>>>      *
> >>>>>>>      2.
> >>>>>>> *
> >>>>>>>
> >>>>>> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
> >>>>>>>      <
> >>>>>>>
> >>>>>> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
> >>>>>>>      *
> >>>>>>>
> >>>>>>>
> >>>>>>> Best～
> >>>>>>>
> >>>>>>>
> >>>>>>>
> >>>>>>> Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
> >>>>>>>
> >>>>>>>> Hi,
> >>>>>>>>
> >>>>>>>> The current rescale API appears to be a work in progress. A couple
> >>>>>> years
> >>>>>>>> ago, we disabled access to the API [1].
> >>>>>>>>
> >>>>>>>> I'm looking into this problem as part of working on autoscaling [2]
> >>>>>> where
> >>>>>>>> we currently require a full restart of the job to apply the parallelism
> >>>>>>>> overrides. This adds additional delay and comes with the caveat that we
> >>>>>>>> don't know whether sufficient resources are available prior to
> >>>>>> executing
> >>>>>>>> the scaling decision. We obviously do not want to get stuck due to a
> >>>>>> lack
> >>>>>>>> of resources. So a rescale API would have to ensure enough resources
> >>>>>> are
> >>>>>>>> available prior to restarting the job.
> >>>>>>>>
> >>>>>>>> I've created an issue here:
> >>>>>>>> https://issues.apache.org/jira/browse/FLINK-30773
> >>>>>>>>
> >>>>>>>> Any comments or interest in working on this?
> >>>>>>>>
> >>>>>>>> -Max
> >>>>>>>>
> >>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-12312
> >>>>>>>> [2]
> >>>>>>>>
> >>>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
> >>>>>>> --
> >>>>>>> Best
> >>>>>>>
> >>>>>>> ConradJam
> >>>>>>>
>

Re: Reworking the Rescale API

Posted by Chesnay Schepler <ch...@apache.org>.

> My primary concern here is to be able to rescale with upfront reservation of resources before restarting the job, so the job doesn't get stuck in case of resource constraints.

Not sure I follow. The AS only rescales when it has already acquired the slots that it needs.

 > This is a blocker from my side. Why do we have that restriction?

We just didn't bother fixing it initially. It should be easy to fix.

On 02/02/2023 18:29, Maximilian Michels wrote:
> I fully agree that in-place scaling is a much harder problem which is
> out of the scope for now. My primary concern here is to be able to
> rescale with upfront reservation of resources before restarting the
> job, so the job doesn't get stuck in case of resource constraints.
>
>> Unused slots: If the max parallelism for slot sharing groups is not equal, slots offered to Adaptive Scheduler might be unused.
> From: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/#limitations-1
>
> This is a blocker from my side. Why do we have that restriction?
>
> On Thu, Feb 2, 2023 at 5:03 PM Chesnay Schepler <ch...@apache.org> wrote:
>>   > If I understand correctly, the adaptive scheduler currently does a
>> full job restart. Is there any work planned to enable in-place rescaling
>> with the adaptive scheduler?
>>
>> Nothing concrete. Sure, it's on a wishlist, but it'd require significant
>> changes to how the runtime works.
>> Rescaling stateful operators requires keygroups to be redistributed,
>> you'd need to be able to change task edges dynamically, roll-back to a
>> checkpoint without restarting tasks, ...
>>
>> It's less of a scheduler thing actually.
>>
>> An earlier step to that would be to allow recovery from an error without
>> restarting all tasks, which would benefit all schedulers.
>> But again bit of a moonshot.
>>
>>   > How well has the adaptive scheduler been tested in production? If we
>> are intending to use it for rescale operations, I'm a bit concerned
>> those jobs might show different behavior due to the scheduling than jobs
>> started with the default scheduler.
>>
>> I don't think we got a lot of feedback so far.
>> Outside of the limitations listed on the elastic scaling page (which I
>> believe we'll address in due time) I'm not aware of any problems.
>> We haven't run into any issues internally.
>>
>> On 02/02/2023 12:44, Maximilian Michels wrote:
>>> +1 on improving the scheduler docs.
>>>
>>>> They never shared a base class since day 1. Are you maybe mixing up the AdaptiveScheduler and AdaptiveBatchScheduler?
>>> @Chesnay: Indeed, I had mixed this up. DefaultScheduler and
>>> AdaptiveScheduler only share the SchedulerNG interface while the
>>> DefaultScheduler and the AdaptiveBatchScheduler share a subset of the
>>> code. Too many schedulers :)
>>>
>>> Thanks for clarifying the current and the intended feature set of the
>>> adaptive scheduler!
>>>
>>> How well has the adaptive scheduler been tested in production? If we
>>> are intending to use it for rescale operations, I'm a bit concerned
>>> those jobs might show different behavior due to the scheduling than
>>> jobs started with the default scheduler.
>>>
>>> If I understand correctly, the adaptive scheduler currently does a
>>> full job restart. Is there any work planned to enable in-place
>>> rescaling with the adaptive scheduler?
>>>
>>>> @max:
>>>>     - when user repartition, we still need to restart the job, can we try to
>>>>     do this part of the work internally instead of externally, as
>>>>     *@konstantin* said only trigger rescaling when the checkpoint or
>>>>     retain-checkpoint is completed operations to minimize reprocessing
>>> @ConradJam: I'm not sure I understand your question. Do you mean when
>>> the partition strategy changes between operators? That shouldn't be
>>> the case for Rescale (except maybe converting ForwardPartitioner to
>>> RescalePartitioner). A more advanced rescale API could allow user
>>> control over this but for now I think it would only support adjusting
>>> parallelism of vertices.
>>>
>>> -Max
>>>
>>> On Thu, Feb 2, 2023 at 6:44 AM weijie guo <gu...@gmail.com> wrote:
>>>> Hi David,
>>>>
>>>> Sorry I'm late to join discuss.
>>>>
>>>> +1 for having a more structure doc about scheduler ecosystem and I can help to fill in the details about batch part.
>>>>
>>>> Best regards,
>>>>
>>>> Weijie
>>>>
>>>>
>>>>
>>>> David Morávek <dm...@apache.org> 于2023年2月1日周三 22:38写道：
>>>>> It makes sense to give the whole "scheduler ecosystem," not just the
>>>>> adaptive scheduler, a little bit more structure in the docs. We already
>>>>> have 4 different schedulers (Default, Adaptive, AdaptiveBatch,
>>>>> AdaptiveBatchSpeculative), and it becomes quite confusing since the details
>>>>> are scattered around the docs. Maybe having a "Job Schedulers" subpage, the
>>>>> same way as we have for "Resource Providers" could do the trick.
>>>>>
>>>>> I should be able to fill in the details about the streaming ones, but I
>>>>> will probably need some help with the batch ones.
>>>>>
>>>>> As for the first FLIP, it's already prepared and we should be able to
>>>>> publish it until Friday.
>>>>>
>>>>> Best,
>>>>> D.
>>>>>
>>>>>
>>>>> On Wed, Feb 1, 2023 at 9:56 AM Gyula Fóra <gy...@gmail.com> wrote:
>>>>>
>>>>>> Chesnay, David:
>>>>>>
>>>>>> Thank you guys for the extra information. We were clearly missing some
>>>>>> context here around the scheduler related efforts and the currently
>>>>>> available feature set.
>>>>>>
>>>>>> As for the concrete suggestions regarding the docs.
>>>>>>
>>>>>> 1. If the adaptive scheduler provides a significantly different feature set
>>>>>> from the default scheduler we could have its own smaller doc page detailing
>>>>>> the differences and why people should switch to it for streaming. This will
>>>>>> also help us when we are making the transition and change the default
>>>>>> behaviour.
>>>>>> 2. We could still have an elastic scaling page that links to the adaptive
>>>>>> scheduler (and vice versa) that focuses on elastic scaling + the Kubernetes
>>>>>> operator autoscaler for a complete picture on elastic scaling options +
>>>>>> detailing the limitations of the different approaches.
>>>>>>
>>>>>> This way the Adaptive Scheduler docs will be decoupled from elastic scaling
>>>>>> and will result in a better understanding for the users (it sure would have
>>>>>> helped us here, and we are on the more advanced user side :))
>>>>>>
>>>>>> What do you think?
>>>>>> Gyula
>>>>>>
>>>>>> On Sat, Jan 28, 2023 at 4:20 AM ConradJam <ja...@gmail.com> wrote:
>>>>>>
>>>>>>> Sorry I'm late to join discuss, I've gleaned a lot of useful information
>>>>>>> from you guys
>>>>>>>
>>>>>>> *@max*
>>>>>>>
>>>>>>>      - when user repartition, we still need to restart the job, can we try
>>>>>> to
>>>>>>>      do this part of the work internally instead of externally, as
>>>>>>>      *@konstantin* said only trigger rescaling when the checkpoint or
>>>>>>>      retain-checkpoint is completed operations to minimize reprocessing
>>>>>>>
>>>>>>> *@konstantin*
>>>>>>>
>>>>>>>      - I think you mentioned that 2 FLIPs are being drafted which I
>>>>>> consider
>>>>>>>      to be the condition to achieve the *@max* goal, I would love to join
>>>>>>>      this discussion and contribute it. I've tried a native implementation
>>>>>> of
>>>>>>>      this part myself, if I can help the community that's the best I can do
>>>>>>>
>>>>>>> *@chesnay*
>>>>>>>
>>>>>>>      - The docs section is confusion/misconceptions confusing like *@gyula
>>>>>>> *say,
>>>>>>>      I'll see if I can fix it
>>>>>>>
>>>>>>>
>>>>>>> *About Rescale Api*
>>>>>>>
>>>>>>>     Some limitations and differences between *default* and *reactive mode*
>>>>>>> were
>>>>>>> discussed earlier, and *@chesnay* explained some of their limitations and
>>>>>>> behaviors, essentially they are two different things. I agree that when
>>>>>>> reactive mode is ready, it should be used as the *reactive mode* for the
>>>>>>> default *stream processing* job.
>>>>>>>     As for the *[1] **Rescale API*, as we know now it seems to be
>>>>>> unusable, I
>>>>>>> believe the goal of this api is to be able to do fast reparallelism. I
>>>>>>> would like to wait until the discussion is over and the 2 draft FILPs
>>>>>>> mentioned earlier are completed. It is not too late to make another
>>>>>>> decision on whether to modify the *[2] **Rescale Rest API *to support for
>>>>>>> parallelism modification of job vertices
>>>>>>>
>>>>>>>
>>>>>>>      1.
>>>>>>> *
>>>>>>>
>>>>>> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
>>>>>>>      <
>>>>>>>
>>>>>> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
>>>>>>>      *
>>>>>>>      2.
>>>>>>> *
>>>>>>>
>>>>>> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
>>>>>>>      <
>>>>>>>
>>>>>> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
>>>>>>>      *
>>>>>>>
>>>>>>>
>>>>>>> Best～
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
>>>>>>>
>>>>>>>> Hi,
>>>>>>>>
>>>>>>>> The current rescale API appears to be a work in progress. A couple
>>>>>> years
>>>>>>>> ago, we disabled access to the API [1].
>>>>>>>>
>>>>>>>> I'm looking into this problem as part of working on autoscaling [2]
>>>>>> where
>>>>>>>> we currently require a full restart of the job to apply the parallelism
>>>>>>>> overrides. This adds additional delay and comes with the caveat that we
>>>>>>>> don't know whether sufficient resources are available prior to
>>>>>> executing
>>>>>>>> the scaling decision. We obviously do not want to get stuck due to a
>>>>>> lack
>>>>>>>> of resources. So a rescale API would have to ensure enough resources
>>>>>> are
>>>>>>>> available prior to restarting the job.
>>>>>>>>
>>>>>>>> I've created an issue here:
>>>>>>>> https://issues.apache.org/jira/browse/FLINK-30773
>>>>>>>>
>>>>>>>> Any comments or interest in working on this?
>>>>>>>>
>>>>>>>> -Max
>>>>>>>>
>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-12312
>>>>>>>> [2]
>>>>>>>>
>>>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
>>>>>>> --
>>>>>>> Best
>>>>>>>
>>>>>>> ConradJam
>>>>>>>

Re: Reworking the Rescale API

Posted by Maximilian Michels <mx...@apache.org>.

I fully agree that in-place scaling is a much harder problem which is
out of the scope for now. My primary concern here is to be able to
rescale with upfront reservation of resources before restarting the
job, so the job doesn't get stuck in case of resource constraints.

> Unused slots: If the max parallelism for slot sharing groups is not equal, slots offered to Adaptive Scheduler might be unused.
From: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/#limitations-1

This is a blocker from my side. Why do we have that restriction?

On Thu, Feb 2, 2023 at 5:03 PM Chesnay Schepler <ch...@apache.org> wrote:
>
>  > If I understand correctly, the adaptive scheduler currently does a
> full job restart. Is there any work planned to enable in-place rescaling
> with the adaptive scheduler?
>
> Nothing concrete. Sure, it's on a wishlist, but it'd require significant
> changes to how the runtime works.
> Rescaling stateful operators requires keygroups to be redistributed,
> you'd need to be able to change task edges dynamically, roll-back to a
> checkpoint without restarting tasks, ...
>
> It's less of a scheduler thing actually.
>
> An earlier step to that would be to allow recovery from an error without
> restarting all tasks, which would benefit all schedulers.
> But again bit of a moonshot.
>
>  > How well has the adaptive scheduler been tested in production? If we
> are intending to use it for rescale operations, I'm a bit concerned
> those jobs might show different behavior due to the scheduling than jobs
> started with the default scheduler.
>
> I don't think we got a lot of feedback so far.
> Outside of the limitations listed on the elastic scaling page (which I
> believe we'll address in due time) I'm not aware of any problems.
> We haven't run into any issues internally.
>
> On 02/02/2023 12:44, Maximilian Michels wrote:
> > +1 on improving the scheduler docs.
> >
> >> They never shared a base class since day 1. Are you maybe mixing up the AdaptiveScheduler and AdaptiveBatchScheduler?
> > @Chesnay: Indeed, I had mixed this up. DefaultScheduler and
> > AdaptiveScheduler only share the SchedulerNG interface while the
> > DefaultScheduler and the AdaptiveBatchScheduler share a subset of the
> > code. Too many schedulers :)
> >
> > Thanks for clarifying the current and the intended feature set of the
> > adaptive scheduler!
> >
> > How well has the adaptive scheduler been tested in production? If we
> > are intending to use it for rescale operations, I'm a bit concerned
> > those jobs might show different behavior due to the scheduling than
> > jobs started with the default scheduler.
> >
> > If I understand correctly, the adaptive scheduler currently does a
> > full job restart. Is there any work planned to enable in-place
> > rescaling with the adaptive scheduler?
> >
> >> @max:
> >>    - when user repartition, we still need to restart the job, can we try to
> >>    do this part of the work internally instead of externally, as
> >>    *@konstantin* said only trigger rescaling when the checkpoint or
> >>    retain-checkpoint is completed operations to minimize reprocessing
> > @ConradJam: I'm not sure I understand your question. Do you mean when
> > the partition strategy changes between operators? That shouldn't be
> > the case for Rescale (except maybe converting ForwardPartitioner to
> > RescalePartitioner). A more advanced rescale API could allow user
> > control over this but for now I think it would only support adjusting
> > parallelism of vertices.
> >
> > -Max
> >
> > On Thu, Feb 2, 2023 at 6:44 AM weijie guo <gu...@gmail.com> wrote:
> >> Hi David,
> >>
> >> Sorry I'm late to join discuss.
> >>
> >> +1 for having a more structure doc about scheduler ecosystem and I can help to fill in the details about batch part.
> >>
> >> Best regards,
> >>
> >> Weijie
> >>
> >>
> >>
> >> David Morávek <dm...@apache.org> 于2023年2月1日周三 22:38写道：
> >>> It makes sense to give the whole "scheduler ecosystem," not just the
> >>> adaptive scheduler, a little bit more structure in the docs. We already
> >>> have 4 different schedulers (Default, Adaptive, AdaptiveBatch,
> >>> AdaptiveBatchSpeculative), and it becomes quite confusing since the details
> >>> are scattered around the docs. Maybe having a "Job Schedulers" subpage, the
> >>> same way as we have for "Resource Providers" could do the trick.
> >>>
> >>> I should be able to fill in the details about the streaming ones, but I
> >>> will probably need some help with the batch ones.
> >>>
> >>> As for the first FLIP, it's already prepared and we should be able to
> >>> publish it until Friday.
> >>>
> >>> Best,
> >>> D.
> >>>
> >>>
> >>> On Wed, Feb 1, 2023 at 9:56 AM Gyula Fóra <gy...@gmail.com> wrote:
> >>>
> >>>> Chesnay, David:
> >>>>
> >>>> Thank you guys for the extra information. We were clearly missing some
> >>>> context here around the scheduler related efforts and the currently
> >>>> available feature set.
> >>>>
> >>>> As for the concrete suggestions regarding the docs.
> >>>>
> >>>> 1. If the adaptive scheduler provides a significantly different feature set
> >>>> from the default scheduler we could have its own smaller doc page detailing
> >>>> the differences and why people should switch to it for streaming. This will
> >>>> also help us when we are making the transition and change the default
> >>>> behaviour.
> >>>> 2. We could still have an elastic scaling page that links to the adaptive
> >>>> scheduler (and vice versa) that focuses on elastic scaling + the Kubernetes
> >>>> operator autoscaler for a complete picture on elastic scaling options +
> >>>> detailing the limitations of the different approaches.
> >>>>
> >>>> This way the Adaptive Scheduler docs will be decoupled from elastic scaling
> >>>> and will result in a better understanding for the users (it sure would have
> >>>> helped us here, and we are on the more advanced user side :))
> >>>>
> >>>> What do you think?
> >>>> Gyula
> >>>>
> >>>> On Sat, Jan 28, 2023 at 4:20 AM ConradJam <ja...@gmail.com> wrote:
> >>>>
> >>>>> Sorry I'm late to join discuss, I've gleaned a lot of useful information
> >>>>> from you guys
> >>>>>
> >>>>> *@max*
> >>>>>
> >>>>>     - when user repartition, we still need to restart the job, can we try
> >>>> to
> >>>>>     do this part of the work internally instead of externally, as
> >>>>>     *@konstantin* said only trigger rescaling when the checkpoint or
> >>>>>     retain-checkpoint is completed operations to minimize reprocessing
> >>>>>
> >>>>> *@konstantin*
> >>>>>
> >>>>>     - I think you mentioned that 2 FLIPs are being drafted which I
> >>>> consider
> >>>>>     to be the condition to achieve the *@max* goal, I would love to join
> >>>>>     this discussion and contribute it. I've tried a native implementation
> >>>> of
> >>>>>     this part myself, if I can help the community that's the best I can do
> >>>>>
> >>>>> *@chesnay*
> >>>>>
> >>>>>     - The docs section is confusion/misconceptions confusing like *@gyula
> >>>>> *say,
> >>>>>     I'll see if I can fix it
> >>>>>
> >>>>>
> >>>>> *About Rescale Api*
> >>>>>
> >>>>>    Some limitations and differences between *default* and *reactive mode*
> >>>>> were
> >>>>> discussed earlier, and *@chesnay* explained some of their limitations and
> >>>>> behaviors, essentially they are two different things. I agree that when
> >>>>> reactive mode is ready, it should be used as the *reactive mode* for the
> >>>>> default *stream processing* job.
> >>>>>    As for the *[1] **Rescale API*, as we know now it seems to be
> >>>> unusable, I
> >>>>> believe the goal of this api is to be able to do fast reparallelism. I
> >>>>> would like to wait until the discussion is over and the 2 draft FILPs
> >>>>> mentioned earlier are completed. It is not too late to make another
> >>>>> decision on whether to modify the *[2] **Rescale Rest API *to support for
> >>>>> parallelism modification of job vertices
> >>>>>
> >>>>>
> >>>>>     1.
> >>>>> *
> >>>>>
> >>>> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
> >>>>>     <
> >>>>>
> >>>> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
> >>>>>     *
> >>>>>     2.
> >>>>> *
> >>>>>
> >>>> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
> >>>>>     <
> >>>>>
> >>>> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
> >>>>>     *
> >>>>>
> >>>>>
> >>>>> Best～
> >>>>>
> >>>>>
> >>>>>
> >>>>> Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
> >>>>>
> >>>>>> Hi,
> >>>>>>
> >>>>>> The current rescale API appears to be a work in progress. A couple
> >>>> years
> >>>>>> ago, we disabled access to the API [1].
> >>>>>>
> >>>>>> I'm looking into this problem as part of working on autoscaling [2]
> >>>> where
> >>>>>> we currently require a full restart of the job to apply the parallelism
> >>>>>> overrides. This adds additional delay and comes with the caveat that we
> >>>>>> don't know whether sufficient resources are available prior to
> >>>> executing
> >>>>>> the scaling decision. We obviously do not want to get stuck due to a
> >>>> lack
> >>>>>> of resources. So a rescale API would have to ensure enough resources
> >>>> are
> >>>>>> available prior to restarting the job.
> >>>>>>
> >>>>>> I've created an issue here:
> >>>>>> https://issues.apache.org/jira/browse/FLINK-30773
> >>>>>>
> >>>>>> Any comments or interest in working on this?
> >>>>>>
> >>>>>> -Max
> >>>>>>
> >>>>>> [1] https://issues.apache.org/jira/browse/FLINK-12312
> >>>>>> [2]
> >>>>>>
> >>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
> >>>>>
> >>>>> --
> >>>>> Best
> >>>>>
> >>>>> ConradJam
> >>>>>
>

Re: Reworking the Rescale API

Posted by Chesnay Schepler <ch...@apache.org>.

 > If I understand correctly, the adaptive scheduler currently does a 
full job restart. Is there any work planned to enable in-place rescaling 
with the adaptive scheduler?

Nothing concrete. Sure, it's on a wishlist, but it'd require significant 
changes to how the runtime works.
Rescaling stateful operators requires keygroups to be redistributed, 
you'd need to be able to change task edges dynamically, roll-back to a 
checkpoint without restarting tasks, ...

It's less of a scheduler thing actually.

An earlier step to that would be to allow recovery from an error without 
restarting all tasks, which would benefit all schedulers.
But again bit of a moonshot.

 > How well has the adaptive scheduler been tested in production? If we 
are intending to use it for rescale operations, I'm a bit concerned 
those jobs might show different behavior due to the scheduling than jobs 
started with the default scheduler.

I don't think we got a lot of feedback so far.
Outside of the limitations listed on the elastic scaling page (which I 
believe we'll address in due time) I'm not aware of any problems.
We haven't run into any issues internally.

On 02/02/2023 12:44, Maximilian Michels wrote:
> +1 on improving the scheduler docs.
>
>> They never shared a base class since day 1. Are you maybe mixing up the AdaptiveScheduler and AdaptiveBatchScheduler?
> @Chesnay: Indeed, I had mixed this up. DefaultScheduler and
> AdaptiveScheduler only share the SchedulerNG interface while the
> DefaultScheduler and the AdaptiveBatchScheduler share a subset of the
> code. Too many schedulers :)
>
> Thanks for clarifying the current and the intended feature set of the
> adaptive scheduler!
>
> How well has the adaptive scheduler been tested in production? If we
> are intending to use it for rescale operations, I'm a bit concerned
> those jobs might show different behavior due to the scheduling than
> jobs started with the default scheduler.
>
> If I understand correctly, the adaptive scheduler currently does a
> full job restart. Is there any work planned to enable in-place
> rescaling with the adaptive scheduler?
>
>> @max:
>>    - when user repartition, we still need to restart the job, can we try to
>>    do this part of the work internally instead of externally, as
>>    *@konstantin* said only trigger rescaling when the checkpoint or
>>    retain-checkpoint is completed operations to minimize reprocessing
> @ConradJam: I'm not sure I understand your question. Do you mean when
> the partition strategy changes between operators? That shouldn't be
> the case for Rescale (except maybe converting ForwardPartitioner to
> RescalePartitioner). A more advanced rescale API could allow user
> control over this but for now I think it would only support adjusting
> parallelism of vertices.
>
> -Max
>
> On Thu, Feb 2, 2023 at 6:44 AM weijie guo <gu...@gmail.com> wrote:
>> Hi David,
>>
>> Sorry I'm late to join discuss.
>>
>> +1 for having a more structure doc about scheduler ecosystem and I can help to fill in the details about batch part.
>>
>> Best regards,
>>
>> Weijie
>>
>>
>>
>> David Morávek <dm...@apache.org> 于2023年2月1日周三 22:38写道：
>>> It makes sense to give the whole "scheduler ecosystem," not just the
>>> adaptive scheduler, a little bit more structure in the docs. We already
>>> have 4 different schedulers (Default, Adaptive, AdaptiveBatch,
>>> AdaptiveBatchSpeculative), and it becomes quite confusing since the details
>>> are scattered around the docs. Maybe having a "Job Schedulers" subpage, the
>>> same way as we have for "Resource Providers" could do the trick.
>>>
>>> I should be able to fill in the details about the streaming ones, but I
>>> will probably need some help with the batch ones.
>>>
>>> As for the first FLIP, it's already prepared and we should be able to
>>> publish it until Friday.
>>>
>>> Best,
>>> D.
>>>
>>>
>>> On Wed, Feb 1, 2023 at 9:56 AM Gyula Fóra <gy...@gmail.com> wrote:
>>>
>>>> Chesnay, David:
>>>>
>>>> Thank you guys for the extra information. We were clearly missing some
>>>> context here around the scheduler related efforts and the currently
>>>> available feature set.
>>>>
>>>> As for the concrete suggestions regarding the docs.
>>>>
>>>> 1. If the adaptive scheduler provides a significantly different feature set
>>>> from the default scheduler we could have its own smaller doc page detailing
>>>> the differences and why people should switch to it for streaming. This will
>>>> also help us when we are making the transition and change the default
>>>> behaviour.
>>>> 2. We could still have an elastic scaling page that links to the adaptive
>>>> scheduler (and vice versa) that focuses on elastic scaling + the Kubernetes
>>>> operator autoscaler for a complete picture on elastic scaling options +
>>>> detailing the limitations of the different approaches.
>>>>
>>>> This way the Adaptive Scheduler docs will be decoupled from elastic scaling
>>>> and will result in a better understanding for the users (it sure would have
>>>> helped us here, and we are on the more advanced user side :))
>>>>
>>>> What do you think?
>>>> Gyula
>>>>
>>>> On Sat, Jan 28, 2023 at 4:20 AM ConradJam <ja...@gmail.com> wrote:
>>>>
>>>>> Sorry I'm late to join discuss, I've gleaned a lot of useful information
>>>>> from you guys
>>>>>
>>>>> *@max*
>>>>>
>>>>>     - when user repartition, we still need to restart the job, can we try
>>>> to
>>>>>     do this part of the work internally instead of externally, as
>>>>>     *@konstantin* said only trigger rescaling when the checkpoint or
>>>>>     retain-checkpoint is completed operations to minimize reprocessing
>>>>>
>>>>> *@konstantin*
>>>>>
>>>>>     - I think you mentioned that 2 FLIPs are being drafted which I
>>>> consider
>>>>>     to be the condition to achieve the *@max* goal, I would love to join
>>>>>     this discussion and contribute it. I've tried a native implementation
>>>> of
>>>>>     this part myself, if I can help the community that's the best I can do
>>>>>
>>>>> *@chesnay*
>>>>>
>>>>>     - The docs section is confusion/misconceptions confusing like *@gyula
>>>>> *say,
>>>>>     I'll see if I can fix it
>>>>>
>>>>>
>>>>> *About Rescale Api*
>>>>>
>>>>>    Some limitations and differences between *default* and *reactive mode*
>>>>> were
>>>>> discussed earlier, and *@chesnay* explained some of their limitations and
>>>>> behaviors, essentially they are two different things. I agree that when
>>>>> reactive mode is ready, it should be used as the *reactive mode* for the
>>>>> default *stream processing* job.
>>>>>    As for the *[1] **Rescale API*, as we know now it seems to be
>>>> unusable, I
>>>>> believe the goal of this api is to be able to do fast reparallelism. I
>>>>> would like to wait until the discussion is over and the 2 draft FILPs
>>>>> mentioned earlier are completed. It is not too late to make another
>>>>> decision on whether to modify the *[2] **Rescale Rest API *to support for
>>>>> parallelism modification of job vertices
>>>>>
>>>>>
>>>>>     1.
>>>>> *
>>>>>
>>>> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
>>>>>     <
>>>>>
>>>> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
>>>>>     *
>>>>>     2.
>>>>> *
>>>>>
>>>> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
>>>>>     <
>>>>>
>>>> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
>>>>>     *
>>>>>
>>>>>
>>>>> Best～
>>>>>
>>>>>
>>>>>
>>>>> Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> The current rescale API appears to be a work in progress. A couple
>>>> years
>>>>>> ago, we disabled access to the API [1].
>>>>>>
>>>>>> I'm looking into this problem as part of working on autoscaling [2]
>>>> where
>>>>>> we currently require a full restart of the job to apply the parallelism
>>>>>> overrides. This adds additional delay and comes with the caveat that we
>>>>>> don't know whether sufficient resources are available prior to
>>>> executing
>>>>>> the scaling decision. We obviously do not want to get stuck due to a
>>>> lack
>>>>>> of resources. So a rescale API would have to ensure enough resources
>>>> are
>>>>>> available prior to restarting the job.
>>>>>>
>>>>>> I've created an issue here:
>>>>>> https://issues.apache.org/jira/browse/FLINK-30773
>>>>>>
>>>>>> Any comments or interest in working on this?
>>>>>>
>>>>>> -Max
>>>>>>
>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-12312
>>>>>> [2]
>>>>>>
>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
>>>>>
>>>>> --
>>>>> Best
>>>>>
>>>>> ConradJam
>>>>>

Re: Reworking the Rescale API

Posted by Maximilian Michels <mx...@apache.org>.

+1 on improving the scheduler docs.

> They never shared a base class since day 1. Are you maybe mixing up the AdaptiveScheduler and AdaptiveBatchScheduler?

@Chesnay: Indeed, I had mixed this up. DefaultScheduler and
AdaptiveScheduler only share the SchedulerNG interface while the
DefaultScheduler and the AdaptiveBatchScheduler share a subset of the
code. Too many schedulers :)

Thanks for clarifying the current and the intended feature set of the
adaptive scheduler!

How well has the adaptive scheduler been tested in production? If we
are intending to use it for rescale operations, I'm a bit concerned
those jobs might show different behavior due to the scheduling than
jobs started with the default scheduler.

If I understand correctly, the adaptive scheduler currently does a
full job restart. Is there any work planned to enable in-place
rescaling with the adaptive scheduler?

> @max:
>   - when user repartition, we still need to restart the job, can we try to
>   do this part of the work internally instead of externally, as
>   *@konstantin* said only trigger rescaling when the checkpoint or
>   retain-checkpoint is completed operations to minimize reprocessing

@ConradJam: I'm not sure I understand your question. Do you mean when
the partition strategy changes between operators? That shouldn't be
the case for Rescale (except maybe converting ForwardPartitioner to
RescalePartitioner). A more advanced rescale API could allow user
control over this but for now I think it would only support adjusting
parallelism of vertices.

-Max

On Thu, Feb 2, 2023 at 6:44 AM weijie guo <gu...@gmail.com> wrote:
>
> Hi David,
>
> Sorry I'm late to join discuss.
>
> +1 for having a more structure doc about scheduler ecosystem and I can help to fill in the details about batch part.
>
> Best regards,
>
> Weijie
>
>
>
> David Morávek <dm...@apache.org> 于2023年2月1日周三 22:38写道：
>>
>> It makes sense to give the whole "scheduler ecosystem," not just the
>> adaptive scheduler, a little bit more structure in the docs. We already
>> have 4 different schedulers (Default, Adaptive, AdaptiveBatch,
>> AdaptiveBatchSpeculative), and it becomes quite confusing since the details
>> are scattered around the docs. Maybe having a "Job Schedulers" subpage, the
>> same way as we have for "Resource Providers" could do the trick.
>>
>> I should be able to fill in the details about the streaming ones, but I
>> will probably need some help with the batch ones.
>>
>> As for the first FLIP, it's already prepared and we should be able to
>> publish it until Friday.
>>
>> Best,
>> D.
>>
>>
>> On Wed, Feb 1, 2023 at 9:56 AM Gyula Fóra <gy...@gmail.com> wrote:
>>
>> > Chesnay, David:
>> >
>> > Thank you guys for the extra information. We were clearly missing some
>> > context here around the scheduler related efforts and the currently
>> > available feature set.
>> >
>> > As for the concrete suggestions regarding the docs.
>> >
>> > 1. If the adaptive scheduler provides a significantly different feature set
>> > from the default scheduler we could have its own smaller doc page detailing
>> > the differences and why people should switch to it for streaming. This will
>> > also help us when we are making the transition and change the default
>> > behaviour.
>> > 2. We could still have an elastic scaling page that links to the adaptive
>> > scheduler (and vice versa) that focuses on elastic scaling + the Kubernetes
>> > operator autoscaler for a complete picture on elastic scaling options +
>> > detailing the limitations of the different approaches.
>> >
>> > This way the Adaptive Scheduler docs will be decoupled from elastic scaling
>> > and will result in a better understanding for the users (it sure would have
>> > helped us here, and we are on the more advanced user side :))
>> >
>> > What do you think?
>> > Gyula
>> >
>> > On Sat, Jan 28, 2023 at 4:20 AM ConradJam <ja...@gmail.com> wrote:
>> >
>> > > Sorry I'm late to join discuss, I've gleaned a lot of useful information
>> > > from you guys
>> > >
>> > > *@max*
>> > >
>> > >    - when user repartition, we still need to restart the job, can we try
>> > to
>> > >    do this part of the work internally instead of externally, as
>> > >    *@konstantin* said only trigger rescaling when the checkpoint or
>> > >    retain-checkpoint is completed operations to minimize reprocessing
>> > >
>> > > *@konstantin*
>> > >
>> > >    - I think you mentioned that 2 FLIPs are being drafted which I
>> > consider
>> > >    to be the condition to achieve the *@max* goal, I would love to join
>> > >    this discussion and contribute it. I've tried a native implementation
>> > of
>> > >    this part myself, if I can help the community that's the best I can do
>> > >
>> > > *@chesnay*
>> > >
>> > >    - The docs section is confusion/misconceptions confusing like *@gyula
>> > > *say,
>> > >    I'll see if I can fix it
>> > >
>> > >
>> > > *About Rescale Api*
>> > >
>> > >   Some limitations and differences between *default* and *reactive mode*
>> > > were
>> > > discussed earlier, and *@chesnay* explained some of their limitations and
>> > > behaviors, essentially they are two different things. I agree that when
>> > > reactive mode is ready, it should be used as the *reactive mode* for the
>> > > default *stream processing* job.
>> > >   As for the *[1] **Rescale API*, as we know now it seems to be
>> > unusable, I
>> > > believe the goal of this api is to be able to do fast reparallelism. I
>> > > would like to wait until the discussion is over and the 2 draft FILPs
>> > > mentioned earlier are completed. It is not too late to make another
>> > > decision on whether to modify the *[2] **Rescale Rest API *to support for
>> > > parallelism modification of job vertices
>> > >
>> > >
>> > >    1.
>> > > *
>> > >
>> > https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
>> > >    <
>> > >
>> > https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
>> > > >
>> > >    *
>> > >    2.
>> > > *
>> > >
>> > https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
>> > >    <
>> > >
>> > https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
>> > > >
>> > >    *
>> > >
>> > >
>> > > Best～
>> > >
>> > >
>> > >
>> > > Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
>> > >
>> > > > Hi,
>> > > >
>> > > > The current rescale API appears to be a work in progress. A couple
>> > years
>> > > > ago, we disabled access to the API [1].
>> > > >
>> > > > I'm looking into this problem as part of working on autoscaling [2]
>> > where
>> > > > we currently require a full restart of the job to apply the parallelism
>> > > > overrides. This adds additional delay and comes with the caveat that we
>> > > > don't know whether sufficient resources are available prior to
>> > executing
>> > > > the scaling decision. We obviously do not want to get stuck due to a
>> > lack
>> > > > of resources. So a rescale API would have to ensure enough resources
>> > are
>> > > > available prior to restarting the job.
>> > > >
>> > > > I've created an issue here:
>> > > > https://issues.apache.org/jira/browse/FLINK-30773
>> > > >
>> > > > Any comments or interest in working on this?
>> > > >
>> > > > -Max
>> > > >
>> > > > [1] https://issues.apache.org/jira/browse/FLINK-12312
>> > > > [2]
>> > > >
>> > >
>> > https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
>> > > >
>> > >
>> > >
>> > > --
>> > > Best
>> > >
>> > > ConradJam
>> > >
>> >

Re: Reworking the Rescale API

Posted by weijie guo <gu...@gmail.com>.

Hi David,

Sorry I'm late to join discuss.

+1 for having a more structure doc about scheduler ecosystem and I can help
to fill in the details about batch part.

Best regards,

Weijie


David Morávek <dm...@apache.org> 于2023年2月1日周三 22:38写道：

> It makes sense to give the whole "scheduler ecosystem," not just the
> adaptive scheduler, a little bit more structure in the docs. We already
> have 4 different schedulers (Default, Adaptive, AdaptiveBatch,
> AdaptiveBatchSpeculative), and it becomes quite confusing since the details
> are scattered around the docs. Maybe having a "Job Schedulers" subpage, the
> same way as we have for "Resource Providers" could do the trick.
>
> I should be able to fill in the details about the streaming ones, but I
> will probably need some help with the batch ones.
>
> As for the first FLIP, it's already prepared and we should be able to
> publish it until Friday.
>
> Best,
> D.
>
>
> On Wed, Feb 1, 2023 at 9:56 AM Gyula Fóra <gy...@gmail.com> wrote:
>
> > Chesnay, David:
> >
> > Thank you guys for the extra information. We were clearly missing some
> > context here around the scheduler related efforts and the currently
> > available feature set.
> >
> > As for the concrete suggestions regarding the docs.
> >
> > 1. If the adaptive scheduler provides a significantly different feature
> set
> > from the default scheduler we could have its own smaller doc page
> detailing
> > the differences and why people should switch to it for streaming. This
> will
> > also help us when we are making the transition and change the default
> > behaviour.
> > 2. We could still have an elastic scaling page that links to the adaptive
> > scheduler (and vice versa) that focuses on elastic scaling + the
> Kubernetes
> > operator autoscaler for a complete picture on elastic scaling options +
> > detailing the limitations of the different approaches.
> >
> > This way the Adaptive Scheduler docs will be decoupled from elastic
> scaling
> > and will result in a better understanding for the users (it sure would
> have
> > helped us here, and we are on the more advanced user side :))
> >
> > What do you think?
> > Gyula
> >
> > On Sat, Jan 28, 2023 at 4:20 AM ConradJam <ja...@gmail.com> wrote:
> >
> > > Sorry I'm late to join discuss, I've gleaned a lot of useful
> information
> > > from you guys
> > >
> > > *@max*
> > >
> > >    - when user repartition, we still need to restart the job, can we
> try
> > to
> > >    do this part of the work internally instead of externally, as
> > >    *@konstantin* said only trigger rescaling when the checkpoint or
> > >    retain-checkpoint is completed operations to minimize reprocessing
> > >
> > > *@konstantin*
> > >
> > >    - I think you mentioned that 2 FLIPs are being drafted which I
> > consider
> > >    to be the condition to achieve the *@max* goal, I would love to join
> > >    this discussion and contribute it. I've tried a native
> implementation
> > of
> > >    this part myself, if I can help the community that's the best I can
> do
> > >
> > > *@chesnay*
> > >
> > >    - The docs section is confusion/misconceptions confusing like
> *@gyula
> > > *say,
> > >    I'll see if I can fix it
> > >
> > >
> > > *About Rescale Api*
> > >
> > >   Some limitations and differences between *default* and *reactive
> mode*
> > > were
> > > discussed earlier, and *@chesnay* explained some of their limitations
> and
> > > behaviors, essentially they are two different things. I agree that when
> > > reactive mode is ready, it should be used as the *reactive mode* for
> the
> > > default *stream processing* job.
> > >   As for the *[1] **Rescale API*, as we know now it seems to be
> > unusable, I
> > > believe the goal of this api is to be able to do fast reparallelism. I
> > > would like to wait until the discussion is over and the 2 draft FILPs
> > > mentioned earlier are completed. It is not too late to make another
> > > decision on whether to modify the *[2] **Rescale Rest API *to support
> for
> > > parallelism modification of job vertices
> > >
> > >
> > >    1.
> > > *
> > >
> >
> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
> > >    <
> > >
> >
> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
> > > >
> > >    *
> > >    2.
> > > *
> > >
> >
> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
> > >    <
> > >
> >
> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
> > > >
> > >    *
> > >
> > >
> > > Best～
> > >
> > >
> > >
> > > Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
> > >
> > > > Hi,
> > > >
> > > > The current rescale API appears to be a work in progress. A couple
> > years
> > > > ago, we disabled access to the API [1].
> > > >
> > > > I'm looking into this problem as part of working on autoscaling [2]
> > where
> > > > we currently require a full restart of the job to apply the
> parallelism
> > > > overrides. This adds additional delay and comes with the caveat that
> we
> > > > don't know whether sufficient resources are available prior to
> > executing
> > > > the scaling decision. We obviously do not want to get stuck due to a
> > lack
> > > > of resources. So a rescale API would have to ensure enough resources
> > are
> > > > available prior to restarting the job.
> > > >
> > > > I've created an issue here:
> > > > https://issues.apache.org/jira/browse/FLINK-30773
> > > >
> > > > Any comments or interest in working on this?
> > > >
> > > > -Max
> > > >
> > > > [1] https://issues.apache.org/jira/browse/FLINK-12312
> > > > [2]
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
> > > >
> > >
> > >
> > > --
> > > Best
> > >
> > > ConradJam
> > >
> >
>

Re: Reworking the Rescale API

Posted by David Morávek <dm...@apache.org>.

It makes sense to give the whole "scheduler ecosystem," not just the
adaptive scheduler, a little bit more structure in the docs. We already
have 4 different schedulers (Default, Adaptive, AdaptiveBatch,
AdaptiveBatchSpeculative), and it becomes quite confusing since the details
are scattered around the docs. Maybe having a "Job Schedulers" subpage, the
same way as we have for "Resource Providers" could do the trick.

I should be able to fill in the details about the streaming ones, but I
will probably need some help with the batch ones.

As for the first FLIP, it's already prepared and we should be able to
publish it until Friday.

Best,
D.


On Wed, Feb 1, 2023 at 9:56 AM Gyula Fóra <gy...@gmail.com> wrote:

> Chesnay, David:
>
> Thank you guys for the extra information. We were clearly missing some
> context here around the scheduler related efforts and the currently
> available feature set.
>
> As for the concrete suggestions regarding the docs.
>
> 1. If the adaptive scheduler provides a significantly different feature set
> from the default scheduler we could have its own smaller doc page detailing
> the differences and why people should switch to it for streaming. This will
> also help us when we are making the transition and change the default
> behaviour.
> 2. We could still have an elastic scaling page that links to the adaptive
> scheduler (and vice versa) that focuses on elastic scaling + the Kubernetes
> operator autoscaler for a complete picture on elastic scaling options +
> detailing the limitations of the different approaches.
>
> This way the Adaptive Scheduler docs will be decoupled from elastic scaling
> and will result in a better understanding for the users (it sure would have
> helped us here, and we are on the more advanced user side :))
>
> What do you think?
> Gyula
>
> On Sat, Jan 28, 2023 at 4:20 AM ConradJam <ja...@gmail.com> wrote:
>
> > Sorry I'm late to join discuss, I've gleaned a lot of useful information
> > from you guys
> >
> > *@max*
> >
> >    - when user repartition, we still need to restart the job, can we try
> to
> >    do this part of the work internally instead of externally, as
> >    *@konstantin* said only trigger rescaling when the checkpoint or
> >    retain-checkpoint is completed operations to minimize reprocessing
> >
> > *@konstantin*
> >
> >    - I think you mentioned that 2 FLIPs are being drafted which I
> consider
> >    to be the condition to achieve the *@max* goal, I would love to join
> >    this discussion and contribute it. I've tried a native implementation
> of
> >    this part myself, if I can help the community that's the best I can do
> >
> > *@chesnay*
> >
> >    - The docs section is confusion/misconceptions confusing like *@gyula
> > *say,
> >    I'll see if I can fix it
> >
> >
> > *About Rescale Api*
> >
> >   Some limitations and differences between *default* and *reactive mode*
> > were
> > discussed earlier, and *@chesnay* explained some of their limitations and
> > behaviors, essentially they are two different things. I agree that when
> > reactive mode is ready, it should be used as the *reactive mode* for the
> > default *stream processing* job.
> >   As for the *[1] **Rescale API*, as we know now it seems to be
> unusable, I
> > believe the goal of this api is to be able to do fast reparallelism. I
> > would like to wait until the discussion is over and the 2 draft FILPs
> > mentioned earlier are completed. It is not too late to make another
> > decision on whether to modify the *[2] **Rescale Rest API *to support for
> > parallelism modification of job vertices
> >
> >
> >    1.
> > *
> >
> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
> >    <
> >
> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
> > >
> >    *
> >    2.
> > *
> >
> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
> >    <
> >
> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
> > >
> >    *
> >
> >
> > Best～
> >
> >
> >
> > Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
> >
> > > Hi,
> > >
> > > The current rescale API appears to be a work in progress. A couple
> years
> > > ago, we disabled access to the API [1].
> > >
> > > I'm looking into this problem as part of working on autoscaling [2]
> where
> > > we currently require a full restart of the job to apply the parallelism
> > > overrides. This adds additional delay and comes with the caveat that we
> > > don't know whether sufficient resources are available prior to
> executing
> > > the scaling decision. We obviously do not want to get stuck due to a
> lack
> > > of resources. So a rescale API would have to ensure enough resources
> are
> > > available prior to restarting the job.
> > >
> > > I've created an issue here:
> > > https://issues.apache.org/jira/browse/FLINK-30773
> > >
> > > Any comments or interest in working on this?
> > >
> > > -Max
> > >
> > > [1] https://issues.apache.org/jira/browse/FLINK-12312
> > > [2]
> > >
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
> > >
> >
> >
> > --
> > Best
> >
> > ConradJam
> >
>

Re: Reworking the Rescale API

Posted by Gyula Fóra <gy...@gmail.com>.

Chesnay, David:

Thank you guys for the extra information. We were clearly missing some
context here around the scheduler related efforts and the currently
available feature set.

As for the concrete suggestions regarding the docs.

1. If the adaptive scheduler provides a significantly different feature set
from the default scheduler we could have its own smaller doc page detailing
the differences and why people should switch to it for streaming. This will
also help us when we are making the transition and change the default
behaviour.
2. We could still have an elastic scaling page that links to the adaptive
scheduler (and vice versa) that focuses on elastic scaling + the Kubernetes
operator autoscaler for a complete picture on elastic scaling options +
detailing the limitations of the different approaches.

This way the Adaptive Scheduler docs will be decoupled from elastic scaling
and will result in a better understanding for the users (it sure would have
helped us here, and we are on the more advanced user side :))

What do you think?
Gyula

On Sat, Jan 28, 2023 at 4:20 AM ConradJam <ja...@gmail.com> wrote:

> Sorry I'm late to join discuss, I've gleaned a lot of useful information
> from you guys
>
> *@max*
>
>    - when user repartition, we still need to restart the job, can we try to
>    do this part of the work internally instead of externally, as
>    *@konstantin* said only trigger rescaling when the checkpoint or
>    retain-checkpoint is completed operations to minimize reprocessing
>
> *@konstantin*
>
>    - I think you mentioned that 2 FLIPs are being drafted which I consider
>    to be the condition to achieve the *@max* goal, I would love to join
>    this discussion and contribute it. I've tried a native implementation of
>    this part myself, if I can help the community that's the best I can do
>
> *@chesnay*
>
>    - The docs section is confusion/misconceptions confusing like *@gyula
> *say,
>    I'll see if I can fix it
>
>
> *About Rescale Api*
>
>   Some limitations and differences between *default* and *reactive mode*
> were
> discussed earlier, and *@chesnay* explained some of their limitations and
> behaviors, essentially they are two different things. I agree that when
> reactive mode is ready, it should be used as the *reactive mode* for the
> default *stream processing* job.
>   As for the *[1] **Rescale API*, as we know now it seems to be unusable, I
> believe the goal of this api is to be able to do fast reparallelism. I
> would like to wait until the discussion is over and the 2 draft FILPs
> mentioned earlier are completed. It is not too late to make another
> decision on whether to modify the *[2] **Rescale Rest API *to support for
> parallelism modification of job vertices
>
>
>    1.
> *
> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
>    <
> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
> >
>    *
>    2.
> *
> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
>    <
> https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
> >
>    *
>
>
> Best～
>
>
>
> Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
>
> > Hi,
> >
> > The current rescale API appears to be a work in progress. A couple years
> > ago, we disabled access to the API [1].
> >
> > I'm looking into this problem as part of working on autoscaling [2] where
> > we currently require a full restart of the job to apply the parallelism
> > overrides. This adds additional delay and comes with the caveat that we
> > don't know whether sufficient resources are available prior to executing
> > the scaling decision. We obviously do not want to get stuck due to a lack
> > of resources. So a rescale API would have to ensure enough resources are
> > available prior to restarting the job.
> >
> > I've created an issue here:
> > https://issues.apache.org/jira/browse/FLINK-30773
> >
> > Any comments or interest in working on this?
> >
> > -Max
> >
> > [1] https://issues.apache.org/jira/browse/FLINK-12312
> > [2]
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
> >
>
>
> --
> Best
>
> ConradJam
>

Re: Reworking the Rescale API

Posted by ConradJam <ja...@gmail.com>.

Sorry I'm late to join discuss, I've gleaned a lot of useful information
from you guys

*@max*

   - when user repartition, we still need to restart the job, can we try to
   do this part of the work internally instead of externally, as
   *@konstantin* said only trigger rescaling when the checkpoint or
   retain-checkpoint is completed operations to minimize reprocessing

*@konstantin*

   - I think you mentioned that 2 FLIPs are being drafted which I consider
   to be the condition to achieve the *@max* goal, I would love to join
   this discussion and contribute it. I've tried a native implementation of
   this part myself, if I can help the community that's the best I can do

*@chesnay*

   - The docs section is confusion/misconceptions confusing like *@gyula *say,
   I'll see if I can fix it


*About Rescale Api*

  Some limitations and differences between *default* and *reactive mode* were
discussed earlier, and *@chesnay* explained some of their limitations and
behaviors, essentially they are two different things. I agree that when
reactive mode is ready, it should be used as the *reactive mode* for the
default *stream processing* job.
  As for the *[1] **Rescale API*, as we know now it seems to be unusable, I
believe the goal of this api is to be able to do fast reparallelism. I
would like to wait until the discussion is over and the 2 draft FILPs
mentioned earlier are completed. It is not too late to make another
decision on whether to modify the *[2] **Rescale Rest API *to support for
parallelism modification of job vertices


   1.
*https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
   <https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/>
   *
   2.
*https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling
   <https://nightlies.apache.org/flink/flink-docs-master/docs/ops/rest_api/#jobs-jobid-rescaling>
   *


Best～



Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：

> Hi,
>
> The current rescale API appears to be a work in progress. A couple years
> ago, we disabled access to the API [1].
>
> I'm looking into this problem as part of working on autoscaling [2] where
> we currently require a full restart of the job to apply the parallelism
> overrides. This adds additional delay and comes with the caveat that we
> don't know whether sufficient resources are available prior to executing
> the scaling decision. We obviously do not want to get stuck due to a lack
> of resources. So a rescale API would have to ensure enough resources are
> available prior to restarting the job.
>
> I've created an issue here:
> https://issues.apache.org/jira/browse/FLINK-30773
>
> Any comments or interest in working on this?
>
> -Max
>
> [1] https://issues.apache.org/jira/browse/FLINK-12312
> [2]
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
>


-- 
Best

ConradJam

Re: Reworking the Rescale API

Posted by Gyula Fóra <gy...@gmail.com>.

Thank you @Chesnay Schepler <ch...@apache.org> @David Morávek
<dm...@apache.org>

I think in that case our primary goal should be to make sure that streaming
jobs always use the adaptive scheduler.
Also then it makes perfect sense to build the rescale api improvements for
that specifically.

However we should have a clear plan to make this a default and communicate
to the users. Otherwise this will only cause further confusion to the users
if we add more and more adaptive scheduler specific features.

Cheers,
Gyula

On Fri, Jan 27, 2023 at 9:59 AM Chesnay Schepler <ch...@apache.org> wrote:

> The adaptive scheduler only supports streaming jobs. That's the biggest
> limitation that probably won't be fixed anytime soon.
> The goal was though to make the adaptive scheduler the default for
> streaming jobs eventually.
> it was very much meant as a better version of the default scheduler for
> streaming jobs.
>
> On 26/01/2023 19:06, David Morávek wrote:
> > Hi Gyula,
> >
> >
> >> can you please explain why the AdaptiveScheduler is not the default
> >> scheduler?
> >
> > There are still some smaller bits missing. As far as I know, the missing
> > parts are:
> >
> > 1) Local recovery (reusing the already downloaded state files after
> restart
> > / rescale)
> > 2) Support for fine-grained resource management
> > 3) Support for the session cluster (Chesnay will be submitting a FLIP for
> > this soon)
> >
> > We're looking into addressing all of these limitations in the short term.
> >
> > Personally, I'd love to start a discussion about making transitioning the
> > AdaptiveScheduler into a default one after those limitations are fixed.
> > Being able to eventually deprecate and remove the DefaultScheduler would
> > simplify the code-base by a lot since there are many adapters between new
> > and old interfaces (eg. SlotPool-related interfaces).
> >
> > Best,
> > D.
> >
> > On Thu, Jan 26, 2023 at 6:27 PM Gyula Fóra <gy...@gmail.com> wrote:
> >
> >> Chesnay,
> >>
> >> Seems like you are suggesting that the Adaptive scheduler does
> everything
> >> the standard scheduler does and more.
> >>
> >> I am clearly not an expert on this topic but can you please explain why
> the
> >> AdaptiveScheduler is not the default scheduler?
> >> If it can do everything, why do we even have 2 schedulers? Why not
> simply
> >> drop the "old" one?
> >>
> >> That would probably clear up all confusionsthen :)
> >>
> >> Gyula
> >>
> >> On Thu, Jan 26, 2023 at 6:23 PM Chesnay Schepler <ch...@apache.org>
> >> wrote:
> >>
> >>> There's the default and reactive mode; nothing else.
> >>> At it's core they are the same thing; reactive mode just cranks up the
> >>> desired parallelism to infinity and enforces certain assumptions (e.g.,
> >>> no active resource management).
> >>>
> >>> The advantage is that the adaptive scheduler can run jobs while not
> >>> sufficient resources are available, and scale things up again once they
> >>> are available.
> >>> This is it's core functionality, but we always intended to extend it
> >>> such that users can modify the parallelism at runtime as well.
> >>> And since the AS can already rescale jobs (and was purpose-built with
> >>> that functionality in mind), this is just a matter of exposing an API
> >>> for it. Everything else is already there.
> >>>
> >>> As a concrete use-case, let's say you have an SLA that says jobs must
> >>> not be down longer than X seconds, and a TM just crashed.
> >>> If you can absolutely guarantee that your k8s cluster can provision a
> >>> new TM within X seconds, no matter what cruel reality has in store for
> >>> you, than you /may/ not need it.
> >>> If you can't, well then here's a use-case for you.
> >>>
> >>>   > Last time I looked they implemented the same interface and the same
> >>> base class. Of course, their behavior is quite different.
> >>>
> >>> They never shared a base class since day 1. Are you maybe mixing up the
> >>> AdaptiveScheduler and AdaptiveBatchScheduler?
> >>>
> >>> As for FLINK-30773, I think that should be covered.
> >>>
> >>> On 26/01/2023 17:10, Maximilian Michels wrote:
> >>>> Thanks for the explanation. If not for the "reactive mode", what is
> >>>> the advantage of the adaptive scheduler? What other modes does it
> >>>> support?
> >>>>
> >>>>> Apart from implementing the same interface the implementations of the
> >>> adaptive and default schedulers are separate.
> >>>> Last time I looked they implemented the same interface and the same
> >>>> base class. Of course, their behavior is quite different.
> >>>>
> >>>> I'm still very interested in learning about the future FLIPs
> >>>> mentioned. Based on the replies, I'm assuming that they will support
> >>>> the changes required for
> >>>> https://issues.apache.org/jira/browse/FLINK-30773, or at least
> provide
> >>>> the basis for implementing them.
> >>>>
> >>>> -Max
> >>>>
> >>>> On Thu, Jan 26, 2023 at 4:57 PM Chesnay Schepler<ch...@apache.org>
> >>> wrote:
> >>>>> On 26/01/2023 16:18, Maximilian Michels wrote:
> >>>>>
> >>>>> I see slightly different goals for the standard and the adaptive
> >>>>> scheduler. The adaptive scheduler's goal is to adapt the Flink job
> >>>>> according to the available resources.
> >>>>>
> >>>>> This is really a misconception that we just have to stomp out.
> >>>>>
> >>>>> This statement only applies to reactive mode, a special mode in which
> >>> the adaptive scheduler (AS) can run in where active resource management
> >> is
> >>> not supported since requesting infinite resources from k8s doesn't
> really
> >>> make sense.
> >>>>> The AS itself can work perfectly fine with active resource
> management,
> >>> and has no effect on how the RM talks to k8s. It can just keep the job
> >>> running in cases where less than desired (==user-provided parallelism)
> >>> resources are provided by k8s (possibly temporarily).
> >>>>> On 26/01/2023 16:18, Maximilian Michels wrote:
> >>>>>
> >>>>> After
> >>>>> all, both schedulers share the same super class
> >>>>>
> >>>>> Apart from implementing the same interface the implementations of the
> >>> adaptive and default schedulers are separate.
> >>>
> >>>
>
>

Re: Reworking the Rescale API

Posted by Chesnay Schepler <ch...@apache.org>.

It is certainly true that the messaging around the AS/reactive mode 
wasn't good.

In part this happened because initially we only intended to advertise 
reactive mode (at the time), and only later figured that the AS on it's 
own could already be useful too.

That being said, I'm not sure how to improve the elastic scaling page, 
can you make specific suggestions?
For example I don't see it being implied that using the adaptive 
schedulers uses reactive mode by default (which it doesn't).
Some tight coupling is of course required because reactive mode requires 
the AS, but to me the page makes it clear that you can use the AS on 
it's own.

The only thing I can think of is making reactive mode a sub-section of 
the AS.
I may have just been too involved to really see the problems in the 
docs; I'd appreciate any help you can give to improve the docs.

On 27/01/2023 10:33, Gyula Fóra wrote:
> Also @David Morávek <dm...@apache.org> @Chesnay Schepler <ch...@apache.org>
>
>
> It would be great if you could update the respective docs page before
> publishing your improvement FLIPS about the adaptive scheduler:
> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/
>
> I think many of the confusion/misconceptions stem from this docs page as it
> ties the Adaptive scheduler and Reactive mode closely together.
> It also clearly implies that if you use the adaptive scheduler, then
> reactive mode is the default behaviour which is probably not what we want
> if we want users to switch over from the current standard scheduler.
>
> We should clearly separate the features of the Adaptive Scheduler from the
> reactive mode which is tailored for very specific use cases and setups
> (standalone only) and requires careful configuration before production use.
> Reactive mode should be an opt-in feature.
>
> Gyula
>
> On Fri, Jan 27, 2023 at 10:19 AM David Morávek <dm...@apache.org> wrote:
>
>>> The adaptive scheduler only supports streaming jobs. That's the biggest
>>> limitation that probably won't be fixed anytime soon.
>>
>> Since FLIP-283 [1] has been accepted, I think this limitation might have
>> already been addressed to a certain extent. I'd be completely fine with
>> having a separate scheduler for batch and streaming (maybe we could build a
>> hybrid one at some point that automatically switches between the two).
>>
>> [1]
>>
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-283%3A+Use+adaptive+batch+scheduler+as+default+scheduler+for+batch+jobs
>>
>>
>> On Fri, Jan 27, 2023 at 9:58 AM Chesnay Schepler <ch...@apache.org>
>> wrote:
>>
>>> The adaptive scheduler only supports streaming jobs. That's the biggest
>>> limitation that probably won't be fixed anytime soon.
>>> The goal was though to make the adaptive scheduler the default for
>>> streaming jobs eventually.
>>> it was very much meant as a better version of the default scheduler for
>>> streaming jobs.
>>>
>>> On 26/01/2023 19:06, David Morávek wrote:
>>>> Hi Gyula,
>>>>
>>>>
>>>>> can you please explain why the AdaptiveScheduler is not the default
>>>>> scheduler?
>>>> There are still some smaller bits missing. As far as I know, the
>> missing
>>>> parts are:
>>>>
>>>> 1) Local recovery (reusing the already downloaded state files after
>>> restart
>>>> / rescale)
>>>> 2) Support for fine-grained resource management
>>>> 3) Support for the session cluster (Chesnay will be submitting a FLIP
>> for
>>>> this soon)
>>>>
>>>> We're looking into addressing all of these limitations in the short
>> term.
>>>> Personally, I'd love to start a discussion about making transitioning
>> the
>>>> AdaptiveScheduler into a default one after those limitations are fixed.
>>>> Being able to eventually deprecate and remove the DefaultScheduler
>> would
>>>> simplify the code-base by a lot since there are many adapters between
>> new
>>>> and old interfaces (eg. SlotPool-related interfaces).
>>>>
>>>> Best,
>>>> D.
>>>>
>>>> On Thu, Jan 26, 2023 at 6:27 PM Gyula Fóra <gy...@gmail.com>
>> wrote:
>>>>> Chesnay,
>>>>>
>>>>> Seems like you are suggesting that the Adaptive scheduler does
>>> everything
>>>>> the standard scheduler does and more.
>>>>>
>>>>> I am clearly not an expert on this topic but can you please explain
>> why
>>> the
>>>>> AdaptiveScheduler is not the default scheduler?
>>>>> If it can do everything, why do we even have 2 schedulers? Why not
>>> simply
>>>>> drop the "old" one?
>>>>>
>>>>> That would probably clear up all confusionsthen :)
>>>>>
>>>>> Gyula
>>>>>
>>>>> On Thu, Jan 26, 2023 at 6:23 PM Chesnay Schepler <ch...@apache.org>
>>>>> wrote:
>>>>>
>>>>>> There's the default and reactive mode; nothing else.
>>>>>> At it's core they are the same thing; reactive mode just cranks up
>> the
>>>>>> desired parallelism to infinity and enforces certain assumptions
>> (e.g.,
>>>>>> no active resource management).
>>>>>>
>>>>>> The advantage is that the adaptive scheduler can run jobs while not
>>>>>> sufficient resources are available, and scale things up again once
>> they
>>>>>> are available.
>>>>>> This is it's core functionality, but we always intended to extend it
>>>>>> such that users can modify the parallelism at runtime as well.
>>>>>> And since the AS can already rescale jobs (and was purpose-built with
>>>>>> that functionality in mind), this is just a matter of exposing an API
>>>>>> for it. Everything else is already there.
>>>>>>
>>>>>> As a concrete use-case, let's say you have an SLA that says jobs must
>>>>>> not be down longer than X seconds, and a TM just crashed.
>>>>>> If you can absolutely guarantee that your k8s cluster can provision a
>>>>>> new TM within X seconds, no matter what cruel reality has in store
>> for
>>>>>> you, than you /may/ not need it.
>>>>>> If you can't, well then here's a use-case for you.
>>>>>>
>>>>>>    > Last time I looked they implemented the same interface and the
>> same
>>>>>> base class. Of course, their behavior is quite different.
>>>>>>
>>>>>> They never shared a base class since day 1. Are you maybe mixing up
>> the
>>>>>> AdaptiveScheduler and AdaptiveBatchScheduler?
>>>>>>
>>>>>> As for FLINK-30773, I think that should be covered.
>>>>>>
>>>>>> On 26/01/2023 17:10, Maximilian Michels wrote:
>>>>>>> Thanks for the explanation. If not for the "reactive mode", what is
>>>>>>> the advantage of the adaptive scheduler? What other modes does it
>>>>>>> support?
>>>>>>>
>>>>>>>> Apart from implementing the same interface the implementations of
>> the
>>>>>> adaptive and default schedulers are separate.
>>>>>>> Last time I looked they implemented the same interface and the same
>>>>>>> base class. Of course, their behavior is quite different.
>>>>>>>
>>>>>>> I'm still very interested in learning about the future FLIPs
>>>>>>> mentioned. Based on the replies, I'm assuming that they will support
>>>>>>> the changes required for
>>>>>>> https://issues.apache.org/jira/browse/FLINK-30773, or at least
>>> provide
>>>>>>> the basis for implementing them.
>>>>>>>
>>>>>>> -Max
>>>>>>>
>>>>>>> On Thu, Jan 26, 2023 at 4:57 PM Chesnay Schepler<chesnay@apache.org
>>>>>> wrote:
>>>>>>>> On 26/01/2023 16:18, Maximilian Michels wrote:
>>>>>>>>
>>>>>>>> I see slightly different goals for the standard and the adaptive
>>>>>>>> scheduler. The adaptive scheduler's goal is to adapt the Flink job
>>>>>>>> according to the available resources.
>>>>>>>>
>>>>>>>> This is really a misconception that we just have to stomp out.
>>>>>>>>
>>>>>>>> This statement only applies to reactive mode, a special mode in
>> which
>>>>>> the adaptive scheduler (AS) can run in where active resource
>> management
>>>>> is
>>>>>> not supported since requesting infinite resources from k8s doesn't
>>> really
>>>>>> make sense.
>>>>>>>> The AS itself can work perfectly fine with active resource
>>> management,
>>>>>> and has no effect on how the RM talks to k8s. It can just keep the
>> job
>>>>>> running in cases where less than desired (==user-provided
>> parallelism)
>>>>>> resources are provided by k8s (possibly temporarily).
>>>>>>>> On 26/01/2023 16:18, Maximilian Michels wrote:
>>>>>>>>
>>>>>>>> After
>>>>>>>> all, both schedulers share the same super class
>>>>>>>>
>>>>>>>> Apart from implementing the same interface the implementations of
>> the
>>>>>> adaptive and default schedulers are separate.
>>>>>>
>>>>>>
>>>

Re: Reworking the Rescale API

Posted by Gyula Fóra <gy...@gmail.com>.

Also @David Morávek <dm...@apache.org> @Chesnay Schepler <ch...@apache.org>


It would be great if you could update the respective docs page before
publishing your improvement FLIPS about the adaptive scheduler:
https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/elastic_scaling/

I think many of the confusion/misconceptions stem from this docs page as it
ties the Adaptive scheduler and Reactive mode closely together.
It also clearly implies that if you use the adaptive scheduler, then
reactive mode is the default behaviour which is probably not what we want
if we want users to switch over from the current standard scheduler.

We should clearly separate the features of the Adaptive Scheduler from the
reactive mode which is tailored for very specific use cases and setups
(standalone only) and requires careful configuration before production use.
Reactive mode should be an opt-in feature.

Gyula

On Fri, Jan 27, 2023 at 10:19 AM David Morávek <dm...@apache.org> wrote:

> >
> > The adaptive scheduler only supports streaming jobs. That's the biggest
> > limitation that probably won't be fixed anytime soon.
>
>
> Since FLIP-283 [1] has been accepted, I think this limitation might have
> already been addressed to a certain extent. I'd be completely fine with
> having a separate scheduler for batch and streaming (maybe we could build a
> hybrid one at some point that automatically switches between the two).
>
> [1]
>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-283%3A+Use+adaptive+batch+scheduler+as+default+scheduler+for+batch+jobs
>
>
> On Fri, Jan 27, 2023 at 9:58 AM Chesnay Schepler <ch...@apache.org>
> wrote:
>
> > The adaptive scheduler only supports streaming jobs. That's the biggest
> > limitation that probably won't be fixed anytime soon.
> > The goal was though to make the adaptive scheduler the default for
> > streaming jobs eventually.
> > it was very much meant as a better version of the default scheduler for
> > streaming jobs.
> >
> > On 26/01/2023 19:06, David Morávek wrote:
> > > Hi Gyula,
> > >
> > >
> > >> can you please explain why the AdaptiveScheduler is not the default
> > >> scheduler?
> > >
> > > There are still some smaller bits missing. As far as I know, the
> missing
> > > parts are:
> > >
> > > 1) Local recovery (reusing the already downloaded state files after
> > restart
> > > / rescale)
> > > 2) Support for fine-grained resource management
> > > 3) Support for the session cluster (Chesnay will be submitting a FLIP
> for
> > > this soon)
> > >
> > > We're looking into addressing all of these limitations in the short
> term.
> > >
> > > Personally, I'd love to start a discussion about making transitioning
> the
> > > AdaptiveScheduler into a default one after those limitations are fixed.
> > > Being able to eventually deprecate and remove the DefaultScheduler
> would
> > > simplify the code-base by a lot since there are many adapters between
> new
> > > and old interfaces (eg. SlotPool-related interfaces).
> > >
> > > Best,
> > > D.
> > >
> > > On Thu, Jan 26, 2023 at 6:27 PM Gyula Fóra <gy...@gmail.com>
> wrote:
> > >
> > >> Chesnay,
> > >>
> > >> Seems like you are suggesting that the Adaptive scheduler does
> > everything
> > >> the standard scheduler does and more.
> > >>
> > >> I am clearly not an expert on this topic but can you please explain
> why
> > the
> > >> AdaptiveScheduler is not the default scheduler?
> > >> If it can do everything, why do we even have 2 schedulers? Why not
> > simply
> > >> drop the "old" one?
> > >>
> > >> That would probably clear up all confusionsthen :)
> > >>
> > >> Gyula
> > >>
> > >> On Thu, Jan 26, 2023 at 6:23 PM Chesnay Schepler <ch...@apache.org>
> > >> wrote:
> > >>
> > >>> There's the default and reactive mode; nothing else.
> > >>> At it's core they are the same thing; reactive mode just cranks up
> the
> > >>> desired parallelism to infinity and enforces certain assumptions
> (e.g.,
> > >>> no active resource management).
> > >>>
> > >>> The advantage is that the adaptive scheduler can run jobs while not
> > >>> sufficient resources are available, and scale things up again once
> they
> > >>> are available.
> > >>> This is it's core functionality, but we always intended to extend it
> > >>> such that users can modify the parallelism at runtime as well.
> > >>> And since the AS can already rescale jobs (and was purpose-built with
> > >>> that functionality in mind), this is just a matter of exposing an API
> > >>> for it. Everything else is already there.
> > >>>
> > >>> As a concrete use-case, let's say you have an SLA that says jobs must
> > >>> not be down longer than X seconds, and a TM just crashed.
> > >>> If you can absolutely guarantee that your k8s cluster can provision a
> > >>> new TM within X seconds, no matter what cruel reality has in store
> for
> > >>> you, than you /may/ not need it.
> > >>> If you can't, well then here's a use-case for you.
> > >>>
> > >>>   > Last time I looked they implemented the same interface and the
> same
> > >>> base class. Of course, their behavior is quite different.
> > >>>
> > >>> They never shared a base class since day 1. Are you maybe mixing up
> the
> > >>> AdaptiveScheduler and AdaptiveBatchScheduler?
> > >>>
> > >>> As for FLINK-30773, I think that should be covered.
> > >>>
> > >>> On 26/01/2023 17:10, Maximilian Michels wrote:
> > >>>> Thanks for the explanation. If not for the "reactive mode", what is
> > >>>> the advantage of the adaptive scheduler? What other modes does it
> > >>>> support?
> > >>>>
> > >>>>> Apart from implementing the same interface the implementations of
> the
> > >>> adaptive and default schedulers are separate.
> > >>>> Last time I looked they implemented the same interface and the same
> > >>>> base class. Of course, their behavior is quite different.
> > >>>>
> > >>>> I'm still very interested in learning about the future FLIPs
> > >>>> mentioned. Based on the replies, I'm assuming that they will support
> > >>>> the changes required for
> > >>>> https://issues.apache.org/jira/browse/FLINK-30773, or at least
> > provide
> > >>>> the basis for implementing them.
> > >>>>
> > >>>> -Max
> > >>>>
> > >>>> On Thu, Jan 26, 2023 at 4:57 PM Chesnay Schepler<chesnay@apache.org
> >
> > >>> wrote:
> > >>>>> On 26/01/2023 16:18, Maximilian Michels wrote:
> > >>>>>
> > >>>>> I see slightly different goals for the standard and the adaptive
> > >>>>> scheduler. The adaptive scheduler's goal is to adapt the Flink job
> > >>>>> according to the available resources.
> > >>>>>
> > >>>>> This is really a misconception that we just have to stomp out.
> > >>>>>
> > >>>>> This statement only applies to reactive mode, a special mode in
> which
> > >>> the adaptive scheduler (AS) can run in where active resource
> management
> > >> is
> > >>> not supported since requesting infinite resources from k8s doesn't
> > really
> > >>> make sense.
> > >>>>> The AS itself can work perfectly fine with active resource
> > management,
> > >>> and has no effect on how the RM talks to k8s. It can just keep the
> job
> > >>> running in cases where less than desired (==user-provided
> parallelism)
> > >>> resources are provided by k8s (possibly temporarily).
> > >>>>> On 26/01/2023 16:18, Maximilian Michels wrote:
> > >>>>>
> > >>>>> After
> > >>>>> all, both schedulers share the same super class
> > >>>>>
> > >>>>> Apart from implementing the same interface the implementations of
> the
> > >>> adaptive and default schedulers are separate.
> > >>>
> > >>>
> >
> >
>

Re: Reworking the Rescale API

Posted by David Morávek <dm...@apache.org>.

>
> The adaptive scheduler only supports streaming jobs. That's the biggest
> limitation that probably won't be fixed anytime soon.


Since FLIP-283 [1] has been accepted, I think this limitation might have
already been addressed to a certain extent. I'd be completely fine with
having a separate scheduler for batch and streaming (maybe we could build a
hybrid one at some point that automatically switches between the two).

[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-283%3A+Use+adaptive+batch+scheduler+as+default+scheduler+for+batch+jobs


On Fri, Jan 27, 2023 at 9:58 AM Chesnay Schepler <ch...@apache.org> wrote:

> The adaptive scheduler only supports streaming jobs. That's the biggest
> limitation that probably won't be fixed anytime soon.
> The goal was though to make the adaptive scheduler the default for
> streaming jobs eventually.
> it was very much meant as a better version of the default scheduler for
> streaming jobs.
>
> On 26/01/2023 19:06, David Morávek wrote:
> > Hi Gyula,
> >
> >
> >> can you please explain why the AdaptiveScheduler is not the default
> >> scheduler?
> >
> > There are still some smaller bits missing. As far as I know, the missing
> > parts are:
> >
> > 1) Local recovery (reusing the already downloaded state files after
> restart
> > / rescale)
> > 2) Support for fine-grained resource management
> > 3) Support for the session cluster (Chesnay will be submitting a FLIP for
> > this soon)
> >
> > We're looking into addressing all of these limitations in the short term.
> >
> > Personally, I'd love to start a discussion about making transitioning the
> > AdaptiveScheduler into a default one after those limitations are fixed.
> > Being able to eventually deprecate and remove the DefaultScheduler would
> > simplify the code-base by a lot since there are many adapters between new
> > and old interfaces (eg. SlotPool-related interfaces).
> >
> > Best,
> > D.
> >
> > On Thu, Jan 26, 2023 at 6:27 PM Gyula Fóra <gy...@gmail.com> wrote:
> >
> >> Chesnay,
> >>
> >> Seems like you are suggesting that the Adaptive scheduler does
> everything
> >> the standard scheduler does and more.
> >>
> >> I am clearly not an expert on this topic but can you please explain why
> the
> >> AdaptiveScheduler is not the default scheduler?
> >> If it can do everything, why do we even have 2 schedulers? Why not
> simply
> >> drop the "old" one?
> >>
> >> That would probably clear up all confusionsthen :)
> >>
> >> Gyula
> >>
> >> On Thu, Jan 26, 2023 at 6:23 PM Chesnay Schepler <ch...@apache.org>
> >> wrote:
> >>
> >>> There's the default and reactive mode; nothing else.
> >>> At it's core they are the same thing; reactive mode just cranks up the
> >>> desired parallelism to infinity and enforces certain assumptions (e.g.,
> >>> no active resource management).
> >>>
> >>> The advantage is that the adaptive scheduler can run jobs while not
> >>> sufficient resources are available, and scale things up again once they
> >>> are available.
> >>> This is it's core functionality, but we always intended to extend it
> >>> such that users can modify the parallelism at runtime as well.
> >>> And since the AS can already rescale jobs (and was purpose-built with
> >>> that functionality in mind), this is just a matter of exposing an API
> >>> for it. Everything else is already there.
> >>>
> >>> As a concrete use-case, let's say you have an SLA that says jobs must
> >>> not be down longer than X seconds, and a TM just crashed.
> >>> If you can absolutely guarantee that your k8s cluster can provision a
> >>> new TM within X seconds, no matter what cruel reality has in store for
> >>> you, than you /may/ not need it.
> >>> If you can't, well then here's a use-case for you.
> >>>
> >>>   > Last time I looked they implemented the same interface and the same
> >>> base class. Of course, their behavior is quite different.
> >>>
> >>> They never shared a base class since day 1. Are you maybe mixing up the
> >>> AdaptiveScheduler and AdaptiveBatchScheduler?
> >>>
> >>> As for FLINK-30773, I think that should be covered.
> >>>
> >>> On 26/01/2023 17:10, Maximilian Michels wrote:
> >>>> Thanks for the explanation. If not for the "reactive mode", what is
> >>>> the advantage of the adaptive scheduler? What other modes does it
> >>>> support?
> >>>>
> >>>>> Apart from implementing the same interface the implementations of the
> >>> adaptive and default schedulers are separate.
> >>>> Last time I looked they implemented the same interface and the same
> >>>> base class. Of course, their behavior is quite different.
> >>>>
> >>>> I'm still very interested in learning about the future FLIPs
> >>>> mentioned. Based on the replies, I'm assuming that they will support
> >>>> the changes required for
> >>>> https://issues.apache.org/jira/browse/FLINK-30773, or at least
> provide
> >>>> the basis for implementing them.
> >>>>
> >>>> -Max
> >>>>
> >>>> On Thu, Jan 26, 2023 at 4:57 PM Chesnay Schepler<ch...@apache.org>
> >>> wrote:
> >>>>> On 26/01/2023 16:18, Maximilian Michels wrote:
> >>>>>
> >>>>> I see slightly different goals for the standard and the adaptive
> >>>>> scheduler. The adaptive scheduler's goal is to adapt the Flink job
> >>>>> according to the available resources.
> >>>>>
> >>>>> This is really a misconception that we just have to stomp out.
> >>>>>
> >>>>> This statement only applies to reactive mode, a special mode in which
> >>> the adaptive scheduler (AS) can run in where active resource management
> >> is
> >>> not supported since requesting infinite resources from k8s doesn't
> really
> >>> make sense.
> >>>>> The AS itself can work perfectly fine with active resource
> management,
> >>> and has no effect on how the RM talks to k8s. It can just keep the job
> >>> running in cases where less than desired (==user-provided parallelism)
> >>> resources are provided by k8s (possibly temporarily).
> >>>>> On 26/01/2023 16:18, Maximilian Michels wrote:
> >>>>>
> >>>>> After
> >>>>> all, both schedulers share the same super class
> >>>>>
> >>>>> Apart from implementing the same interface the implementations of the
> >>> adaptive and default schedulers are separate.
> >>>
> >>>
>
>

Re: Reworking the Rescale API

Posted by Chesnay Schepler <ch...@apache.org>.

The adaptive scheduler only supports streaming jobs. That's the biggest 
limitation that probably won't be fixed anytime soon.
The goal was though to make the adaptive scheduler the default for 
streaming jobs eventually.
it was very much meant as a better version of the default scheduler for 
streaming jobs.

On 26/01/2023 19:06, David Morávek wrote:
> Hi Gyula,
>
>
>> can you please explain why the AdaptiveScheduler is not the default
>> scheduler?
>
> There are still some smaller bits missing. As far as I know, the missing
> parts are:
>
> 1) Local recovery (reusing the already downloaded state files after restart
> / rescale)
> 2) Support for fine-grained resource management
> 3) Support for the session cluster (Chesnay will be submitting a FLIP for
> this soon)
>
> We're looking into addressing all of these limitations in the short term.
>
> Personally, I'd love to start a discussion about making transitioning the
> AdaptiveScheduler into a default one after those limitations are fixed.
> Being able to eventually deprecate and remove the DefaultScheduler would
> simplify the code-base by a lot since there are many adapters between new
> and old interfaces (eg. SlotPool-related interfaces).
>
> Best,
> D.
>
> On Thu, Jan 26, 2023 at 6:27 PM Gyula Fóra <gy...@gmail.com> wrote:
>
>> Chesnay,
>>
>> Seems like you are suggesting that the Adaptive scheduler does everything
>> the standard scheduler does and more.
>>
>> I am clearly not an expert on this topic but can you please explain why the
>> AdaptiveScheduler is not the default scheduler?
>> If it can do everything, why do we even have 2 schedulers? Why not simply
>> drop the "old" one?
>>
>> That would probably clear up all confusionsthen :)
>>
>> Gyula
>>
>> On Thu, Jan 26, 2023 at 6:23 PM Chesnay Schepler <ch...@apache.org>
>> wrote:
>>
>>> There's the default and reactive mode; nothing else.
>>> At it's core they are the same thing; reactive mode just cranks up the
>>> desired parallelism to infinity and enforces certain assumptions (e.g.,
>>> no active resource management).
>>>
>>> The advantage is that the adaptive scheduler can run jobs while not
>>> sufficient resources are available, and scale things up again once they
>>> are available.
>>> This is it's core functionality, but we always intended to extend it
>>> such that users can modify the parallelism at runtime as well.
>>> And since the AS can already rescale jobs (and was purpose-built with
>>> that functionality in mind), this is just a matter of exposing an API
>>> for it. Everything else is already there.
>>>
>>> As a concrete use-case, let's say you have an SLA that says jobs must
>>> not be down longer than X seconds, and a TM just crashed.
>>> If you can absolutely guarantee that your k8s cluster can provision a
>>> new TM within X seconds, no matter what cruel reality has in store for
>>> you, than you /may/ not need it.
>>> If you can't, well then here's a use-case for you.
>>>
>>>   > Last time I looked they implemented the same interface and the same
>>> base class. Of course, their behavior is quite different.
>>>
>>> They never shared a base class since day 1. Are you maybe mixing up the
>>> AdaptiveScheduler and AdaptiveBatchScheduler?
>>>
>>> As for FLINK-30773, I think that should be covered.
>>>
>>> On 26/01/2023 17:10, Maximilian Michels wrote:
>>>> Thanks for the explanation. If not for the "reactive mode", what is
>>>> the advantage of the adaptive scheduler? What other modes does it
>>>> support?
>>>>
>>>>> Apart from implementing the same interface the implementations of the
>>> adaptive and default schedulers are separate.
>>>> Last time I looked they implemented the same interface and the same
>>>> base class. Of course, their behavior is quite different.
>>>>
>>>> I'm still very interested in learning about the future FLIPs
>>>> mentioned. Based on the replies, I'm assuming that they will support
>>>> the changes required for
>>>> https://issues.apache.org/jira/browse/FLINK-30773, or at least provide
>>>> the basis for implementing them.
>>>>
>>>> -Max
>>>>
>>>> On Thu, Jan 26, 2023 at 4:57 PM Chesnay Schepler<ch...@apache.org>
>>> wrote:
>>>>> On 26/01/2023 16:18, Maximilian Michels wrote:
>>>>>
>>>>> I see slightly different goals for the standard and the adaptive
>>>>> scheduler. The adaptive scheduler's goal is to adapt the Flink job
>>>>> according to the available resources.
>>>>>
>>>>> This is really a misconception that we just have to stomp out.
>>>>>
>>>>> This statement only applies to reactive mode, a special mode in which
>>> the adaptive scheduler (AS) can run in where active resource management
>> is
>>> not supported since requesting infinite resources from k8s doesn't really
>>> make sense.
>>>>> The AS itself can work perfectly fine with active resource management,
>>> and has no effect on how the RM talks to k8s. It can just keep the job
>>> running in cases where less than desired (==user-provided parallelism)
>>> resources are provided by k8s (possibly temporarily).
>>>>> On 26/01/2023 16:18, Maximilian Michels wrote:
>>>>>
>>>>> After
>>>>> all, both schedulers share the same super class
>>>>>
>>>>> Apart from implementing the same interface the implementations of the
>>> adaptive and default schedulers are separate.
>>>
>>>

Re: Reworking the Rescale API

Posted by David Morávek <dm...@apache.org>.

Hi Gyula,


> can you please explain why the AdaptiveScheduler is not the default
> scheduler?


There are still some smaller bits missing. As far as I know, the missing
parts are:

1) Local recovery (reusing the already downloaded state files after restart
/ rescale)
2) Support for fine-grained resource management
3) Support for the session cluster (Chesnay will be submitting a FLIP for
this soon)

We're looking into addressing all of these limitations in the short term.

Personally, I'd love to start a discussion about making transitioning the
AdaptiveScheduler into a default one after those limitations are fixed.
Being able to eventually deprecate and remove the DefaultScheduler would
simplify the code-base by a lot since there are many adapters between new
and old interfaces (eg. SlotPool-related interfaces).

Best,
D.

On Thu, Jan 26, 2023 at 6:27 PM Gyula Fóra <gy...@gmail.com> wrote:

> Chesnay,
>
> Seems like you are suggesting that the Adaptive scheduler does everything
> the standard scheduler does and more.
>
> I am clearly not an expert on this topic but can you please explain why the
> AdaptiveScheduler is not the default scheduler?
> If it can do everything, why do we even have 2 schedulers? Why not simply
> drop the "old" one?
>
> That would probably clear up all confusionsthen :)
>
> Gyula
>
> On Thu, Jan 26, 2023 at 6:23 PM Chesnay Schepler <ch...@apache.org>
> wrote:
>
> > There's the default and reactive mode; nothing else.
> > At it's core they are the same thing; reactive mode just cranks up the
> > desired parallelism to infinity and enforces certain assumptions (e.g.,
> > no active resource management).
> >
> > The advantage is that the adaptive scheduler can run jobs while not
> > sufficient resources are available, and scale things up again once they
> > are available.
> > This is it's core functionality, but we always intended to extend it
> > such that users can modify the parallelism at runtime as well.
> > And since the AS can already rescale jobs (and was purpose-built with
> > that functionality in mind), this is just a matter of exposing an API
> > for it. Everything else is already there.
> >
> > As a concrete use-case, let's say you have an SLA that says jobs must
> > not be down longer than X seconds, and a TM just crashed.
> > If you can absolutely guarantee that your k8s cluster can provision a
> > new TM within X seconds, no matter what cruel reality has in store for
> > you, than you /may/ not need it.
> > If you can't, well then here's a use-case for you.
> >
> >  > Last time I looked they implemented the same interface and the same
> > base class. Of course, their behavior is quite different.
> >
> > They never shared a base class since day 1. Are you maybe mixing up the
> > AdaptiveScheduler and AdaptiveBatchScheduler?
> >
> > As for FLINK-30773, I think that should be covered.
> >
> > On 26/01/2023 17:10, Maximilian Michels wrote:
> > > Thanks for the explanation. If not for the "reactive mode", what is
> > > the advantage of the adaptive scheduler? What other modes does it
> > > support?
> > >
> > >> Apart from implementing the same interface the implementations of the
> > adaptive and default schedulers are separate.
> > > Last time I looked they implemented the same interface and the same
> > > base class. Of course, their behavior is quite different.
> > >
> > > I'm still very interested in learning about the future FLIPs
> > > mentioned. Based on the replies, I'm assuming that they will support
> > > the changes required for
> > > https://issues.apache.org/jira/browse/FLINK-30773, or at least provide
> > > the basis for implementing them.
> > >
> > > -Max
> > >
> > > On Thu, Jan 26, 2023 at 4:57 PM Chesnay Schepler<ch...@apache.org>
> > wrote:
> > >> On 26/01/2023 16:18, Maximilian Michels wrote:
> > >>
> > >> I see slightly different goals for the standard and the adaptive
> > >> scheduler. The adaptive scheduler's goal is to adapt the Flink job
> > >> according to the available resources.
> > >>
> > >> This is really a misconception that we just have to stomp out.
> > >>
> > >> This statement only applies to reactive mode, a special mode in which
> > the adaptive scheduler (AS) can run in where active resource management
> is
> > not supported since requesting infinite resources from k8s doesn't really
> > make sense.
> > >>
> > >> The AS itself can work perfectly fine with active resource management,
> > and has no effect on how the RM talks to k8s. It can just keep the job
> > running in cases where less than desired (==user-provided parallelism)
> > resources are provided by k8s (possibly temporarily).
> > >>
> > >> On 26/01/2023 16:18, Maximilian Michels wrote:
> > >>
> > >> After
> > >> all, both schedulers share the same super class
> > >>
> > >> Apart from implementing the same interface the implementations of the
> > adaptive and default schedulers are separate.
> >
> >
>

Re: Reworking the Rescale API

Posted by Gyula Fóra <gy...@gmail.com>.

Chesnay,

Seems like you are suggesting that the Adaptive scheduler does everything
the standard scheduler does and more.

I am clearly not an expert on this topic but can you please explain why the
AdaptiveScheduler is not the default scheduler?
If it can do everything, why do we even have 2 schedulers? Why not simply
drop the "old" one?

That would probably clear up all confusionsthen :)

Gyula

On Thu, Jan 26, 2023 at 6:23 PM Chesnay Schepler <ch...@apache.org> wrote:

> There's the default and reactive mode; nothing else.
> At it's core they are the same thing; reactive mode just cranks up the
> desired parallelism to infinity and enforces certain assumptions (e.g.,
> no active resource management).
>
> The advantage is that the adaptive scheduler can run jobs while not
> sufficient resources are available, and scale things up again once they
> are available.
> This is it's core functionality, but we always intended to extend it
> such that users can modify the parallelism at runtime as well.
> And since the AS can already rescale jobs (and was purpose-built with
> that functionality in mind), this is just a matter of exposing an API
> for it. Everything else is already there.
>
> As a concrete use-case, let's say you have an SLA that says jobs must
> not be down longer than X seconds, and a TM just crashed.
> If you can absolutely guarantee that your k8s cluster can provision a
> new TM within X seconds, no matter what cruel reality has in store for
> you, than you /may/ not need it.
> If you can't, well then here's a use-case for you.
>
>  > Last time I looked they implemented the same interface and the same
> base class. Of course, their behavior is quite different.
>
> They never shared a base class since day 1. Are you maybe mixing up the
> AdaptiveScheduler and AdaptiveBatchScheduler?
>
> As for FLINK-30773, I think that should be covered.
>
> On 26/01/2023 17:10, Maximilian Michels wrote:
> > Thanks for the explanation. If not for the "reactive mode", what is
> > the advantage of the adaptive scheduler? What other modes does it
> > support?
> >
> >> Apart from implementing the same interface the implementations of the
> adaptive and default schedulers are separate.
> > Last time I looked they implemented the same interface and the same
> > base class. Of course, their behavior is quite different.
> >
> > I'm still very interested in learning about the future FLIPs
> > mentioned. Based on the replies, I'm assuming that they will support
> > the changes required for
> > https://issues.apache.org/jira/browse/FLINK-30773, or at least provide
> > the basis for implementing them.
> >
> > -Max
> >
> > On Thu, Jan 26, 2023 at 4:57 PM Chesnay Schepler<ch...@apache.org>
> wrote:
> >> On 26/01/2023 16:18, Maximilian Michels wrote:
> >>
> >> I see slightly different goals for the standard and the adaptive
> >> scheduler. The adaptive scheduler's goal is to adapt the Flink job
> >> according to the available resources.
> >>
> >> This is really a misconception that we just have to stomp out.
> >>
> >> This statement only applies to reactive mode, a special mode in which
> the adaptive scheduler (AS) can run in where active resource management is
> not supported since requesting infinite resources from k8s doesn't really
> make sense.
> >>
> >> The AS itself can work perfectly fine with active resource management,
> and has no effect on how the RM talks to k8s. It can just keep the job
> running in cases where less than desired (==user-provided parallelism)
> resources are provided by k8s (possibly temporarily).
> >>
> >> On 26/01/2023 16:18, Maximilian Michels wrote:
> >>
> >> After
> >> all, both schedulers share the same super class
> >>
> >> Apart from implementing the same interface the implementations of the
> adaptive and default schedulers are separate.
>
>

Re: Reworking the Rescale API

Posted by Chesnay Schepler <ch...@apache.org>.

There's the default and reactive mode; nothing else.
At it's core they are the same thing; reactive mode just cranks up the 
desired parallelism to infinity and enforces certain assumptions (e.g., 
no active resource management).

The advantage is that the adaptive scheduler can run jobs while not 
sufficient resources are available, and scale things up again once they 
are available.
This is it's core functionality, but we always intended to extend it 
such that users can modify the parallelism at runtime as well.
And since the AS can already rescale jobs (and was purpose-built with 
that functionality in mind), this is just a matter of exposing an API 
for it. Everything else is already there.

As a concrete use-case, let's say you have an SLA that says jobs must 
not be down longer than X seconds, and a TM just crashed.
If you can absolutely guarantee that your k8s cluster can provision a 
new TM within X seconds, no matter what cruel reality has in store for 
you, than you /may/ not need it.
If you can't, well then here's a use-case for you.

 > Last time I looked they implemented the same interface and the same 
base class. Of course, their behavior is quite different.

They never shared a base class since day 1. Are you maybe mixing up the 
AdaptiveScheduler and AdaptiveBatchScheduler?

As for FLINK-30773, I think that should be covered.

On 26/01/2023 17:10, Maximilian Michels wrote:
> Thanks for the explanation. If not for the "reactive mode", what is
> the advantage of the adaptive scheduler? What other modes does it
> support?
>
>> Apart from implementing the same interface the implementations of the adaptive and default schedulers are separate.
> Last time I looked they implemented the same interface and the same
> base class. Of course, their behavior is quite different.
>
> I'm still very interested in learning about the future FLIPs
> mentioned. Based on the replies, I'm assuming that they will support
> the changes required for
> https://issues.apache.org/jira/browse/FLINK-30773, or at least provide
> the basis for implementing them.
>
> -Max
>
> On Thu, Jan 26, 2023 at 4:57 PM Chesnay Schepler<ch...@apache.org>  wrote:
>> On 26/01/2023 16:18, Maximilian Michels wrote:
>>
>> I see slightly different goals for the standard and the adaptive
>> scheduler. The adaptive scheduler's goal is to adapt the Flink job
>> according to the available resources.
>>
>> This is really a misconception that we just have to stomp out.
>>
>> This statement only applies to reactive mode, a special mode in which the adaptive scheduler (AS) can run in where active resource management is not supported since requesting infinite resources from k8s doesn't really make sense.
>>
>> The AS itself can work perfectly fine with active resource management, and has no effect on how the RM talks to k8s. It can just keep the job running in cases where less than desired (==user-provided parallelism) resources are provided by k8s (possibly temporarily).
>>
>> On 26/01/2023 16:18, Maximilian Michels wrote:
>>
>> After
>> all, both schedulers share the same super class
>>
>> Apart from implementing the same interface the implementations of the adaptive and default schedulers are separate.

Re: Reworking the Rescale API

Posted by Maximilian Michels <mx...@apache.org>.

Thanks for the explanation. If not for the "reactive mode", what is
the advantage of the adaptive scheduler? What other modes does it
support?

>Apart from implementing the same interface the implementations of the adaptive and default schedulers are separate.

Last time I looked they implemented the same interface and the same
base class. Of course, their behavior is quite different.

I'm still very interested in learning about the future FLIPs
mentioned. Based on the replies, I'm assuming that they will support
the changes required for
https://issues.apache.org/jira/browse/FLINK-30773, or at least provide
the basis for implementing them.

-Max

On Thu, Jan 26, 2023 at 4:57 PM Chesnay Schepler <ch...@apache.org> wrote:
>
> On 26/01/2023 16:18, Maximilian Michels wrote:
>
> I see slightly different goals for the standard and the adaptive
> scheduler. The adaptive scheduler's goal is to adapt the Flink job
> according to the available resources.
>
> This is really a misconception that we just have to stomp out.
>
> This statement only applies to reactive mode, a special mode in which the adaptive scheduler (AS) can run in where active resource management is not supported since requesting infinite resources from k8s doesn't really make sense.
>
> The AS itself can work perfectly fine with active resource management, and has no effect on how the RM talks to k8s. It can just keep the job running in cases where less than desired (==user-provided parallelism) resources are provided by k8s (possibly temporarily).
>
> On 26/01/2023 16:18, Maximilian Michels wrote:
>
> After
> all, both schedulers share the same super class
>
> Apart from implementing the same interface the implementations of the adaptive and default schedulers are separate.

Re: Reworking the Rescale API

Posted by Chesnay Schepler <ch...@apache.org>.

On 26/01/2023 16:18, Maximilian Michels wrote:
> I see slightly different goals for the standard and the adaptive
> scheduler. The adaptive scheduler's goal is to adapt the Flink job
> according to the available resources.

This is really a misconception that we just have to stomp out.

This statement only applies to /reactive mode/, a /special mode/ in 
which the adaptive scheduler (AS) can run in where active resource 
management is not supported since requesting infinite resources from k8s 
doesn't really make sense.

The AS itself can work perfectly fine with active resource management, 
and has no effect on how the RM talks to k8s. It can just keep the job 
running in cases where less than desired (==user-provided parallelism) 
resources are provided by k8s (possibly temporarily).

On 26/01/2023 16:18, Maximilian Michels wrote:
> After
> all, both schedulers share the same super class

Apart from implementing the same interface the implementations of the 
adaptive and default schedulers are separate.

Re: Reworking the Rescale API

Posted by Maximilian Michels <mx...@apache.org>.

Thanks for the replies! I don't mind which scheduler handles the
implementation, as long as autoscaling via the Flink operator works
with it.

I see slightly different goals for the standard and the adaptive
scheduler. The adaptive scheduler's goal is to adapt the Flink job
according to the available resources. The standard scheduler's goal is
to acquire / create as many resources as needed to run the provided
Flink job. From this perspective, the standard scheduler is a better
fit for the Flink Kubernetes operator because it alters only the job
(the operator parallelisms) and leaves the resource management to
Flink.

On the other hand, there are obvious advantages of the adaptive
scheduler, like knowing upfront which resources are available, support
for overriding the parallelism of job vertices, and knowing about
Flink runtime events.

We may be able to adapt the adaptive scheduler to fit the goal of the
Kubernetes operator and provide this functionality through the Rescale
API. How much work that would be, I'm not completely sure yet. After
all, both schedulers share the same super class, so it should be
doable if we add support beyond its current "reactiveness".

>Chesnay and David Moravek are currently in the process of drafting two FLIPs that will extend the support of the adaptive scheduler to session mode and will allow clients to change the desired/min/max parallelism of the vertices of a Job during its runtime via the REST API

It would be good to create a discussion on the mailing list around
these FLIPs you mentioned. This may have already happened, but I
couldn't find anything. As Gyula also mentioned, the session mode is
only one side of the story. It does sound interesting that the
upcoming FLIP may introduce per-vertex parallelism overrides. Maybe
you or the FLIP authors have a good idea how to solve the conflicting
goals between standard and adaptive scheduler (as described above)?

-Max


On Thu, Jan 26, 2023 at 3:19 PM Gyula Fóra <gy...@gmail.com> wrote:
>
> If the adaptive scheduler would support all execution modes like Native
> Applications, Sessions etc including active resource management then I
> think we could use that all the time. I would love to use one scheduler
> instead of having 2 options.
>
> Currently however there is a huge gap in functionality between
> active/passive resource management and from my experience, the active
> (native) integration is much more convenient for Kubernetes environments.
>
> Gyula
>
> On Thu, Jan 26, 2023 at 3:13 PM Konstantin Knauf <kn...@apache.org> wrote:
>
> > Hi Gyula,
> >
> > if the adaptive scheduler supported active resource managers, would there
> > be any other blocker to migrate to it? I don't know much about the
> > implementation-side here, but conceptually once we have session mode
> > support and each Jobs in a session clusters declaris their desired
> > parallelism (!=infinity) there shouldn't be a big gap to support active
> > resource managers. Am I missing something, Chesnay?
> >
> > Regarding the complexity, I was referring to the procedure that Max
> > outlines in his ticket around check if slots are available or not and then
> > triggering scaling operations. The adaptive scheduler already does this and
> > is more responsive in that regard than an external process would be in my
> > understanding.
> >
> > Cheers,
> >
> > Konstantin
> >
> >
> >
> > Am Do., 26. Jan. 2023 um 15:05 Uhr schrieb Gyula Fóra <
> > gyula.fora@gmail.com>:
> >
> >> Hi Konstantin!
> >>
> >> I think the Adaptive Scheduler still will not support Kubernetes Native
> >> integration and can only be used in standalone mode. This means that the
> >> operator needs to manage all resources externally, and compute exactly how
> >> much new slots are needed during rescaling etc.
> >>
> >> I think whatever scaling API we build, it should work for both standalone
> >> and native integration as much as possible. It's not a duplicated effort to
> >> add it to the standard scheduler as long as the adaptive scheduler does not
> >> support active resource management.
> >>
> >> Also it seems this will not reduce complexity on the operator side, which
> >> can already do scaling actions by executing an upgrade.
> >>
> >> And a side note: the operator supports both native and standalone
> >> integration (both standard and adaptive scheduler this way) but the bigger
> >> problem is actually computing the required number of slots and required new
> >> resources which is much harder than simply using active resource management.
> >>
> >> Cheers,
> >> Gyula
> >>
> >> On Thu, Jan 26, 2023 at 2:57 PM Konstantin Knauf <kn...@apache.org>
> >> wrote:
> >>
> >>> Hi Max,
> >>>
> >>> it seems to me we are now running in some of the potential duplication
> >>> of efforts across the standard and adaptive scheduler that Chesnay had
> >>> mentioned on the original ticket. The issue of having to do a full restart
> >>> of the Job for rescaling as well as waiting for resources to be available
> >>> before doing a rescaling operation were some of the main motivations behind
> >>> introducing the adaptive scheduler. In the adaptive scheduler we can
> >>> further do things like only to trigger a rescaling operations exactly when
> >>> a checkpoint was completed to minimize reprocessing. For Jobs with small
> >>> state size, the downtime during rescaling can already be << 1 second today.
> >>>
> >>> Chesnay and David Moravek are currently in the process of drafting two
> >>> FLIPs that will extend the support of the adaptive scheduler to session
> >>> mode and will allow clients to change the desired/min/max parallelism of
> >>> the vertices of a Job during its runtime via the REST API. We currently
> >>> plan to publish a draft of these FLIPs next week for discussion. Would you
> >>> consider moving to the adaptive scheduler for the kubernetes operator
> >>> provided these FLIPs make it? I think, it has the potential to simplify the
> >>> logic required for rescaling on the operator side quite a bit.
> >>>
> >>> Best,
> >>>
> >>> Konstantin
> >>>
> >>>
> >>> Am Do., 26. Jan. 2023 um 12:16 Uhr schrieb Maximilian Michels <
> >>> mxm@apache.org>:
> >>>
> >>>> Hey ConradJam,
> >>>>
> >>>> Thank you for your thoughtful response. It would be great to start
> >>>> writing
> >>>> a FLIP for the Rescale API. If you want to take a stab, please go ahead,
> >>>> I'd be happy to review. I'm sure Gyula or others will also chime in.
> >>>>
> >>>> I want to answer your question so we are aligned:
> >>>>
> >>>> ● Does scaling work on YARN, or just k8s?
> >>>> >
> >>>>
> >>>> I think it should work for both YARN and K8s. We would have to make
> >>>> changes
> >>>> to the drivers (AbstractResourceManagerDriver) which is implemented for
> >>>> both K8s and YARN. The outlined approach for rescaling does not require
> >>>> integrating with those systems, just maybe updating how the driver is
> >>>> used,
> >>>> so we should be able to make it work across both YARN and K8s.
> >>>>
> >>>> ● Rescaling supports Standalone mode?
> >>>> >
> >>>>
> >>>> Yes, I think it should and easily can. We do use a different type of
> >>>> resource manager (StandaloneResourceManager, not ActiveResourceManager)
> >>>> but
> >>>> I think the logic will sit on a higher level where the ResourceManager
> >>>> implementation is not relevant.
> >>>>
> >>>> ● Can we simplify the recovery steps?
> >>>> >
> >>>>
> >>>> For the first version, I would prefer the simple approach of (1)
> >>>> acquiring
> >>>> the required slots for rescaling, then (2) trigger a stop with savepoint
> >>>> (3) resubmit the job with updated parallelisms. What you have in mind
> >>>> is a
> >>>> bit more involved but certainly a great optimization, especially when
> >>>> only
> >>>> a fraction of the job state needs to be repartitioned.
> >>>>
> >>>> Of course, there are many details, such as
> >>>> > ● At some point we may not be able to use this kind of hot update, and
> >>>> > still need to restart the job, when this happens, we should prevent
> >>>> users
> >>>> > from using rescaling requests
> >>>> >
> >>>>
> >>>> I'm curious to learn more about "hot updates". How would we support
> >>>> this in
> >>>> Flink? Would we have to support dynamically repartitioning tasks? I
> >>>> don't
> >>>> think Flink supports this yet. For now, restarting the job may be the
> >>>> best
> >>>> we can do.
> >>>>
> >>>> ● After rescaling is submitted, when we fail, there should be a rollback
> >>>> > mechanism to roll back to the previous degree of parallelism.
> >>>> >
> >>>>
> >>>> This should not be necessary if all the requirements for rescaling, e.g.
> >>>> enough task slots, are satisfied by the Rescale API. I'm not even sure
> >>>> rolling back is an option because we can't guarantee that a rollback
> >>>> would
> >>>> always work.
> >>>>
> >>>> Thanks,
> >>>> Max
> >>>>
> >>>> On Tue, Jan 24, 2023 at 6:34 AM ConradJam <ja...@gmail.com> wrote:
> >>>>
> >>>> > Hello max
> >>>> >
> >>>> > Thanks for driving it, I think there is no problem with your previous
> >>>> > suggestion of [1] FLINK-30773. Here I just put forward some
> >>>> supplements and
> >>>> > doubts.I have some suggestions and insights for this
> >>>> >
> >>>> >  I have experienced the autoscaling of Flink K8S Operator for a part
> >>>> of the
> >>>> > time. The current method is to stop the job and modify the
> >>>> parallelism,
> >>>> > which will interrupt the business for a long time. I think the
> >>>> purpose of
> >>>> > modifying Rescaling Api is to better fit cloud native and reduce the
> >>>> impact
> >>>> > of job scaling downtime.
> >>>> >
> >>>> > I have tried scaling with less time, and I call this step "hot update
> >>>> > parallelism" (if there is an available Slots, there is no need to
> >>>> re-deploy
> >>>> > the JobManager Or TaskManager on K8S)
> >>>> >
> >>>> > Around this topic, I raised the *following questions*:
> >>>> > ● Does scaling work on YARN, or just k8s?
> >>>> >    ○ I think we can support running on K8S for the first version, and
> >>>> Yarn
> >>>> > can be considered later
> >>>> > ● Rescaling supports Standalone mode?
> >>>> >    ○ I think it can be supported. The essence is just to modify the
> >>>> > parallelism of job vertices. As for the tuning strategy, it should be
> >>>> > determined by the external system or K8S Operator
> >>>> > ● Can we simplify the recovery steps?
> >>>> >    ○ As far as I know, the traditional way to adjust the parallelism
> >>>> is to
> >>>> > stop a job and do a Savepoint, and then run the job with the adjusted
> >>>> > parallelism. If we hide this step in the *JobManager*, it will be an
> >>>> > important means to reduce the delay.
> >>>> >
> >>>> >   Of course, there are many details, such as
> >>>> > ● At some point we may not be able to use this kind of hot update, and
> >>>> > still need to restart the job, when this happens, we should prevent
> >>>> users
> >>>> > from using rescaling requests
> >>>> > ● After rescaling is submitted, when we fail, there should be a
> >>>> rollback
> >>>> > mechanism to roll back to the previous degree of parallelism.
> >>>> >
> >>>> > more and more ～
> >>>> >
> >>>> >   By the way, because the content may be more, I did not expand more
> >>>> ideas
> >>>> > and descriptions here. This proposal modifies the original Rescaling
> >>>> API.
> >>>> > I would also like to hear if  *@gyula* has some new ideas on this as
> >>>> it was
> >>>> > also involved in the development of FLIP-271
> >>>> > I am willing to write a FLIP for this purpose to improve and write
> >>>> some
> >>>> > ideas with dev Community and then submit it. What do you think about
> >>>> > starting a discussion for the community?
> >>>> >
> >>>> >
> >>>> >    1. https://issues.apache.org/jira/browse/FLINK-30773
> >>>> >
> >>>> > Best～
> >>>> >
> >>>> > Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
> >>>> >
> >>>> > > Hi,
> >>>> > >
> >>>> > > The current rescale API appears to be a work in progress. A couple
> >>>> years
> >>>> > > ago, we disabled access to the API [1].
> >>>> > >
> >>>> > > I'm looking into this problem as part of working on autoscaling [2]
> >>>> where
> >>>> > > we currently require a full restart of the job to apply the
> >>>> parallelism
> >>>> > > overrides. This adds additional delay and comes with the caveat
> >>>> that we
> >>>> > > don't know whether sufficient resources are available prior to
> >>>> executing
> >>>> > > the scaling decision. We obviously do not want to get stuck due to
> >>>> a lack
> >>>> > > of resources. So a rescale API would have to ensure enough
> >>>> resources are
> >>>> > > available prior to restarting the job.
> >>>> > >
> >>>> > > I've created an issue here:
> >>>> > > https://issues.apache.org/jira/browse/FLINK-30773
> >>>> > >
> >>>> > > Any comments or interest in working on this?
> >>>> > >
> >>>> > > -Max
> >>>> > >
> >>>> > > [1] https://issues.apache.org/jira/browse/FLINK-12312
> >>>> > > [2]
> >>>> > >
> >>>> >
> >>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
> >>>> > >
> >>>> >
> >>>> >
> >>>> > --
> >>>> > Best
> >>>> >
> >>>> > ConradJam
> >>>> >
> >>>> >
> >>>> > --
> >>>> > Best
> >>>> >
> >>>> > ConradJam
> >>>> >
> >>>>
> >>>
> >>>
> >>> --
> >>> https://twitter.com/snntrable
> >>> https://github.com/knaufk
> >>>
> >>
> >
> > --
> > https://twitter.com/snntrable
> > https://github.com/knaufk
> >

Re: Reworking the Rescale API

Posted by Gyula Fóra <gy...@gmail.com>.

If the adaptive scheduler would support all execution modes like Native
Applications, Sessions etc including active resource management then I
think we could use that all the time. I would love to use one scheduler
instead of having 2 options.

Currently however there is a huge gap in functionality between
active/passive resource management and from my experience, the active
(native) integration is much more convenient for Kubernetes environments.

Gyula

On Thu, Jan 26, 2023 at 3:13 PM Konstantin Knauf <kn...@apache.org> wrote:

> Hi Gyula,
>
> if the adaptive scheduler supported active resource managers, would there
> be any other blocker to migrate to it? I don't know much about the
> implementation-side here, but conceptually once we have session mode
> support and each Jobs in a session clusters declaris their desired
> parallelism (!=infinity) there shouldn't be a big gap to support active
> resource managers. Am I missing something, Chesnay?
>
> Regarding the complexity, I was referring to the procedure that Max
> outlines in his ticket around check if slots are available or not and then
> triggering scaling operations. The adaptive scheduler already does this and
> is more responsive in that regard than an external process would be in my
> understanding.
>
> Cheers,
>
> Konstantin
>
>
>
> Am Do., 26. Jan. 2023 um 15:05 Uhr schrieb Gyula Fóra <
> gyula.fora@gmail.com>:
>
>> Hi Konstantin!
>>
>> I think the Adaptive Scheduler still will not support Kubernetes Native
>> integration and can only be used in standalone mode. This means that the
>> operator needs to manage all resources externally, and compute exactly how
>> much new slots are needed during rescaling etc.
>>
>> I think whatever scaling API we build, it should work for both standalone
>> and native integration as much as possible. It's not a duplicated effort to
>> add it to the standard scheduler as long as the adaptive scheduler does not
>> support active resource management.
>>
>> Also it seems this will not reduce complexity on the operator side, which
>> can already do scaling actions by executing an upgrade.
>>
>> And a side note: the operator supports both native and standalone
>> integration (both standard and adaptive scheduler this way) but the bigger
>> problem is actually computing the required number of slots and required new
>> resources which is much harder than simply using active resource management.
>>
>> Cheers,
>> Gyula
>>
>> On Thu, Jan 26, 2023 at 2:57 PM Konstantin Knauf <kn...@apache.org>
>> wrote:
>>
>>> Hi Max,
>>>
>>> it seems to me we are now running in some of the potential duplication
>>> of efforts across the standard and adaptive scheduler that Chesnay had
>>> mentioned on the original ticket. The issue of having to do a full restart
>>> of the Job for rescaling as well as waiting for resources to be available
>>> before doing a rescaling operation were some of the main motivations behind
>>> introducing the adaptive scheduler. In the adaptive scheduler we can
>>> further do things like only to trigger a rescaling operations exactly when
>>> a checkpoint was completed to minimize reprocessing. For Jobs with small
>>> state size, the downtime during rescaling can already be << 1 second today.
>>>
>>> Chesnay and David Moravek are currently in the process of drafting two
>>> FLIPs that will extend the support of the adaptive scheduler to session
>>> mode and will allow clients to change the desired/min/max parallelism of
>>> the vertices of a Job during its runtime via the REST API. We currently
>>> plan to publish a draft of these FLIPs next week for discussion. Would you
>>> consider moving to the adaptive scheduler for the kubernetes operator
>>> provided these FLIPs make it? I think, it has the potential to simplify the
>>> logic required for rescaling on the operator side quite a bit.
>>>
>>> Best,
>>>
>>> Konstantin
>>>
>>>
>>> Am Do., 26. Jan. 2023 um 12:16 Uhr schrieb Maximilian Michels <
>>> mxm@apache.org>:
>>>
>>>> Hey ConradJam,
>>>>
>>>> Thank you for your thoughtful response. It would be great to start
>>>> writing
>>>> a FLIP for the Rescale API. If you want to take a stab, please go ahead,
>>>> I'd be happy to review. I'm sure Gyula or others will also chime in.
>>>>
>>>> I want to answer your question so we are aligned:
>>>>
>>>> ● Does scaling work on YARN, or just k8s?
>>>> >
>>>>
>>>> I think it should work for both YARN and K8s. We would have to make
>>>> changes
>>>> to the drivers (AbstractResourceManagerDriver) which is implemented for
>>>> both K8s and YARN. The outlined approach for rescaling does not require
>>>> integrating with those systems, just maybe updating how the driver is
>>>> used,
>>>> so we should be able to make it work across both YARN and K8s.
>>>>
>>>> ● Rescaling supports Standalone mode?
>>>> >
>>>>
>>>> Yes, I think it should and easily can. We do use a different type of
>>>> resource manager (StandaloneResourceManager, not ActiveResourceManager)
>>>> but
>>>> I think the logic will sit on a higher level where the ResourceManager
>>>> implementation is not relevant.
>>>>
>>>> ● Can we simplify the recovery steps?
>>>> >
>>>>
>>>> For the first version, I would prefer the simple approach of (1)
>>>> acquiring
>>>> the required slots for rescaling, then (2) trigger a stop with savepoint
>>>> (3) resubmit the job with updated parallelisms. What you have in mind
>>>> is a
>>>> bit more involved but certainly a great optimization, especially when
>>>> only
>>>> a fraction of the job state needs to be repartitioned.
>>>>
>>>> Of course, there are many details, such as
>>>> > ● At some point we may not be able to use this kind of hot update, and
>>>> > still need to restart the job, when this happens, we should prevent
>>>> users
>>>> > from using rescaling requests
>>>> >
>>>>
>>>> I'm curious to learn more about "hot updates". How would we support
>>>> this in
>>>> Flink? Would we have to support dynamically repartitioning tasks? I
>>>> don't
>>>> think Flink supports this yet. For now, restarting the job may be the
>>>> best
>>>> we can do.
>>>>
>>>> ● After rescaling is submitted, when we fail, there should be a rollback
>>>> > mechanism to roll back to the previous degree of parallelism.
>>>> >
>>>>
>>>> This should not be necessary if all the requirements for rescaling, e.g.
>>>> enough task slots, are satisfied by the Rescale API. I'm not even sure
>>>> rolling back is an option because we can't guarantee that a rollback
>>>> would
>>>> always work.
>>>>
>>>> Thanks,
>>>> Max
>>>>
>>>> On Tue, Jan 24, 2023 at 6:34 AM ConradJam <ja...@gmail.com> wrote:
>>>>
>>>> > Hello max
>>>> >
>>>> > Thanks for driving it, I think there is no problem with your previous
>>>> > suggestion of [1] FLINK-30773. Here I just put forward some
>>>> supplements and
>>>> > doubts.I have some suggestions and insights for this
>>>> >
>>>> >  I have experienced the autoscaling of Flink K8S Operator for a part
>>>> of the
>>>> > time. The current method is to stop the job and modify the
>>>> parallelism,
>>>> > which will interrupt the business for a long time. I think the
>>>> purpose of
>>>> > modifying Rescaling Api is to better fit cloud native and reduce the
>>>> impact
>>>> > of job scaling downtime.
>>>> >
>>>> > I have tried scaling with less time, and I call this step "hot update
>>>> > parallelism" (if there is an available Slots, there is no need to
>>>> re-deploy
>>>> > the JobManager Or TaskManager on K8S)
>>>> >
>>>> > Around this topic, I raised the *following questions*:
>>>> > ● Does scaling work on YARN, or just k8s?
>>>> >    ○ I think we can support running on K8S for the first version, and
>>>> Yarn
>>>> > can be considered later
>>>> > ● Rescaling supports Standalone mode?
>>>> >    ○ I think it can be supported. The essence is just to modify the
>>>> > parallelism of job vertices. As for the tuning strategy, it should be
>>>> > determined by the external system or K8S Operator
>>>> > ● Can we simplify the recovery steps?
>>>> >    ○ As far as I know, the traditional way to adjust the parallelism
>>>> is to
>>>> > stop a job and do a Savepoint, and then run the job with the adjusted
>>>> > parallelism. If we hide this step in the *JobManager*, it will be an
>>>> > important means to reduce the delay.
>>>> >
>>>> >   Of course, there are many details, such as
>>>> > ● At some point we may not be able to use this kind of hot update, and
>>>> > still need to restart the job, when this happens, we should prevent
>>>> users
>>>> > from using rescaling requests
>>>> > ● After rescaling is submitted, when we fail, there should be a
>>>> rollback
>>>> > mechanism to roll back to the previous degree of parallelism.
>>>> >
>>>> > more and more ～
>>>> >
>>>> >   By the way, because the content may be more, I did not expand more
>>>> ideas
>>>> > and descriptions here. This proposal modifies the original Rescaling
>>>> API.
>>>> > I would also like to hear if  *@gyula* has some new ideas on this as
>>>> it was
>>>> > also involved in the development of FLIP-271
>>>> > I am willing to write a FLIP for this purpose to improve and write
>>>> some
>>>> > ideas with dev Community and then submit it. What do you think about
>>>> > starting a discussion for the community?
>>>> >
>>>> >
>>>> >    1. https://issues.apache.org/jira/browse/FLINK-30773
>>>> >
>>>> > Best～
>>>> >
>>>> > Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
>>>> >
>>>> > > Hi,
>>>> > >
>>>> > > The current rescale API appears to be a work in progress. A couple
>>>> years
>>>> > > ago, we disabled access to the API [1].
>>>> > >
>>>> > > I'm looking into this problem as part of working on autoscaling [2]
>>>> where
>>>> > > we currently require a full restart of the job to apply the
>>>> parallelism
>>>> > > overrides. This adds additional delay and comes with the caveat
>>>> that we
>>>> > > don't know whether sufficient resources are available prior to
>>>> executing
>>>> > > the scaling decision. We obviously do not want to get stuck due to
>>>> a lack
>>>> > > of resources. So a rescale API would have to ensure enough
>>>> resources are
>>>> > > available prior to restarting the job.
>>>> > >
>>>> > > I've created an issue here:
>>>> > > https://issues.apache.org/jira/browse/FLINK-30773
>>>> > >
>>>> > > Any comments or interest in working on this?
>>>> > >
>>>> > > -Max
>>>> > >
>>>> > > [1] https://issues.apache.org/jira/browse/FLINK-12312
>>>> > > [2]
>>>> > >
>>>> >
>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
>>>> > >
>>>> >
>>>> >
>>>> > --
>>>> > Best
>>>> >
>>>> > ConradJam
>>>> >
>>>> >
>>>> > --
>>>> > Best
>>>> >
>>>> > ConradJam
>>>> >
>>>>
>>>
>>>
>>> --
>>> https://twitter.com/snntrable
>>> https://github.com/knaufk
>>>
>>
>
> --
> https://twitter.com/snntrable
> https://github.com/knaufk
>

Re: Reworking the Rescale API

Posted by Konstantin Knauf <kn...@apache.org>.

Hi Gyula,

if the adaptive scheduler supported active resource managers, would there
be any other blocker to migrate to it? I don't know much about the
implementation-side here, but conceptually once we have session mode
support and each Jobs in a session clusters declaris their desired
parallelism (!=infinity) there shouldn't be a big gap to support active
resource managers. Am I missing something, Chesnay?

Regarding the complexity, I was referring to the procedure that Max
outlines in his ticket around check if slots are available or not and then
triggering scaling operations. The adaptive scheduler already does this and
is more responsive in that regard than an external process would be in my
understanding.

Cheers,

Konstantin



Am Do., 26. Jan. 2023 um 15:05 Uhr schrieb Gyula Fóra <gyula.fora@gmail.com
>:

> Hi Konstantin!
>
> I think the Adaptive Scheduler still will not support Kubernetes Native
> integration and can only be used in standalone mode. This means that the
> operator needs to manage all resources externally, and compute exactly how
> much new slots are needed during rescaling etc.
>
> I think whatever scaling API we build, it should work for both standalone
> and native integration as much as possible. It's not a duplicated effort to
> add it to the standard scheduler as long as the adaptive scheduler does not
> support active resource management.
>
> Also it seems this will not reduce complexity on the operator side, which
> can already do scaling actions by executing an upgrade.
>
> And a side note: the operator supports both native and standalone
> integration (both standard and adaptive scheduler this way) but the bigger
> problem is actually computing the required number of slots and required new
> resources which is much harder than simply using active resource management.
>
> Cheers,
> Gyula
>
> On Thu, Jan 26, 2023 at 2:57 PM Konstantin Knauf <kn...@apache.org>
> wrote:
>
>> Hi Max,
>>
>> it seems to me we are now running in some of the potential duplication of
>> efforts across the standard and adaptive scheduler that Chesnay had
>> mentioned on the original ticket. The issue of having to do a full restart
>> of the Job for rescaling as well as waiting for resources to be available
>> before doing a rescaling operation were some of the main motivations behind
>> introducing the adaptive scheduler. In the adaptive scheduler we can
>> further do things like only to trigger a rescaling operations exactly when
>> a checkpoint was completed to minimize reprocessing. For Jobs with small
>> state size, the downtime during rescaling can already be << 1 second today.
>>
>> Chesnay and David Moravek are currently in the process of drafting two
>> FLIPs that will extend the support of the adaptive scheduler to session
>> mode and will allow clients to change the desired/min/max parallelism of
>> the vertices of a Job during its runtime via the REST API. We currently
>> plan to publish a draft of these FLIPs next week for discussion. Would you
>> consider moving to the adaptive scheduler for the kubernetes operator
>> provided these FLIPs make it? I think, it has the potential to simplify the
>> logic required for rescaling on the operator side quite a bit.
>>
>> Best,
>>
>> Konstantin
>>
>>
>> Am Do., 26. Jan. 2023 um 12:16 Uhr schrieb Maximilian Michels <
>> mxm@apache.org>:
>>
>>> Hey ConradJam,
>>>
>>> Thank you for your thoughtful response. It would be great to start
>>> writing
>>> a FLIP for the Rescale API. If you want to take a stab, please go ahead,
>>> I'd be happy to review. I'm sure Gyula or others will also chime in.
>>>
>>> I want to answer your question so we are aligned:
>>>
>>> ● Does scaling work on YARN, or just k8s?
>>> >
>>>
>>> I think it should work for both YARN and K8s. We would have to make
>>> changes
>>> to the drivers (AbstractResourceManagerDriver) which is implemented for
>>> both K8s and YARN. The outlined approach for rescaling does not require
>>> integrating with those systems, just maybe updating how the driver is
>>> used,
>>> so we should be able to make it work across both YARN and K8s.
>>>
>>> ● Rescaling supports Standalone mode?
>>> >
>>>
>>> Yes, I think it should and easily can. We do use a different type of
>>> resource manager (StandaloneResourceManager, not ActiveResourceManager)
>>> but
>>> I think the logic will sit on a higher level where the ResourceManager
>>> implementation is not relevant.
>>>
>>> ● Can we simplify the recovery steps?
>>> >
>>>
>>> For the first version, I would prefer the simple approach of (1)
>>> acquiring
>>> the required slots for rescaling, then (2) trigger a stop with savepoint
>>> (3) resubmit the job with updated parallelisms. What you have in mind is
>>> a
>>> bit more involved but certainly a great optimization, especially when
>>> only
>>> a fraction of the job state needs to be repartitioned.
>>>
>>> Of course, there are many details, such as
>>> > ● At some point we may not be able to use this kind of hot update, and
>>> > still need to restart the job, when this happens, we should prevent
>>> users
>>> > from using rescaling requests
>>> >
>>>
>>> I'm curious to learn more about "hot updates". How would we support this
>>> in
>>> Flink? Would we have to support dynamically repartitioning tasks? I don't
>>> think Flink supports this yet. For now, restarting the job may be the
>>> best
>>> we can do.
>>>
>>> ● After rescaling is submitted, when we fail, there should be a rollback
>>> > mechanism to roll back to the previous degree of parallelism.
>>> >
>>>
>>> This should not be necessary if all the requirements for rescaling, e.g.
>>> enough task slots, are satisfied by the Rescale API. I'm not even sure
>>> rolling back is an option because we can't guarantee that a rollback
>>> would
>>> always work.
>>>
>>> Thanks,
>>> Max
>>>
>>> On Tue, Jan 24, 2023 at 6:34 AM ConradJam <ja...@gmail.com> wrote:
>>>
>>> > Hello max
>>> >
>>> > Thanks for driving it, I think there is no problem with your previous
>>> > suggestion of [1] FLINK-30773. Here I just put forward some
>>> supplements and
>>> > doubts.I have some suggestions and insights for this
>>> >
>>> >  I have experienced the autoscaling of Flink K8S Operator for a part
>>> of the
>>> > time. The current method is to stop the job and modify the parallelism,
>>> > which will interrupt the business for a long time. I think the purpose
>>> of
>>> > modifying Rescaling Api is to better fit cloud native and reduce the
>>> impact
>>> > of job scaling downtime.
>>> >
>>> > I have tried scaling with less time, and I call this step "hot update
>>> > parallelism" (if there is an available Slots, there is no need to
>>> re-deploy
>>> > the JobManager Or TaskManager on K8S)
>>> >
>>> > Around this topic, I raised the *following questions*:
>>> > ● Does scaling work on YARN, or just k8s?
>>> >    ○ I think we can support running on K8S for the first version, and
>>> Yarn
>>> > can be considered later
>>> > ● Rescaling supports Standalone mode?
>>> >    ○ I think it can be supported. The essence is just to modify the
>>> > parallelism of job vertices. As for the tuning strategy, it should be
>>> > determined by the external system or K8S Operator
>>> > ● Can we simplify the recovery steps?
>>> >    ○ As far as I know, the traditional way to adjust the parallelism
>>> is to
>>> > stop a job and do a Savepoint, and then run the job with the adjusted
>>> > parallelism. If we hide this step in the *JobManager*, it will be an
>>> > important means to reduce the delay.
>>> >
>>> >   Of course, there are many details, such as
>>> > ● At some point we may not be able to use this kind of hot update, and
>>> > still need to restart the job, when this happens, we should prevent
>>> users
>>> > from using rescaling requests
>>> > ● After rescaling is submitted, when we fail, there should be a
>>> rollback
>>> > mechanism to roll back to the previous degree of parallelism.
>>> >
>>> > more and more ～
>>> >
>>> >   By the way, because the content may be more, I did not expand more
>>> ideas
>>> > and descriptions here. This proposal modifies the original Rescaling
>>> API.
>>> > I would also like to hear if  *@gyula* has some new ideas on this as
>>> it was
>>> > also involved in the development of FLIP-271
>>> > I am willing to write a FLIP for this purpose to improve and write some
>>> > ideas with dev Community and then submit it. What do you think about
>>> > starting a discussion for the community?
>>> >
>>> >
>>> >    1. https://issues.apache.org/jira/browse/FLINK-30773
>>> >
>>> > Best～
>>> >
>>> > Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
>>> >
>>> > > Hi,
>>> > >
>>> > > The current rescale API appears to be a work in progress. A couple
>>> years
>>> > > ago, we disabled access to the API [1].
>>> > >
>>> > > I'm looking into this problem as part of working on autoscaling [2]
>>> where
>>> > > we currently require a full restart of the job to apply the
>>> parallelism
>>> > > overrides. This adds additional delay and comes with the caveat that
>>> we
>>> > > don't know whether sufficient resources are available prior to
>>> executing
>>> > > the scaling decision. We obviously do not want to get stuck due to a
>>> lack
>>> > > of resources. So a rescale API would have to ensure enough resources
>>> are
>>> > > available prior to restarting the job.
>>> > >
>>> > > I've created an issue here:
>>> > > https://issues.apache.org/jira/browse/FLINK-30773
>>> > >
>>> > > Any comments or interest in working on this?
>>> > >
>>> > > -Max
>>> > >
>>> > > [1] https://issues.apache.org/jira/browse/FLINK-12312
>>> > > [2]
>>> > >
>>> >
>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
>>> > >
>>> >
>>> >
>>> > --
>>> > Best
>>> >
>>> > ConradJam
>>> >
>>> >
>>> > --
>>> > Best
>>> >
>>> > ConradJam
>>> >
>>>
>>
>>
>> --
>> https://twitter.com/snntrable
>> https://github.com/knaufk
>>
>

-- 
https://twitter.com/snntrable
https://github.com/knaufk

Re: Reworking the Rescale API

Posted by Gyula Fóra <gy...@gmail.com>.

Hi Konstantin!

I think the Adaptive Scheduler still will not support Kubernetes Native
integration and can only be used in standalone mode. This means that the
operator needs to manage all resources externally, and compute exactly how
much new slots are needed during rescaling etc.

I think whatever scaling API we build, it should work for both standalone
and native integration as much as possible. It's not a duplicated effort to
add it to the standard scheduler as long as the adaptive scheduler does not
support active resource management.

Also it seems this will not reduce complexity on the operator side, which
can already do scaling actions by executing an upgrade.

And a side note: the operator supports both native and standalone
integration (both standard and adaptive scheduler this way) but the bigger
problem is actually computing the required number of slots and required new
resources which is much harder than simply using active resource management.

Cheers,
Gyula

On Thu, Jan 26, 2023 at 2:57 PM Konstantin Knauf <kn...@apache.org> wrote:

> Hi Max,
>
> it seems to me we are now running in some of the potential duplication of
> efforts across the standard and adaptive scheduler that Chesnay had
> mentioned on the original ticket. The issue of having to do a full restart
> of the Job for rescaling as well as waiting for resources to be available
> before doing a rescaling operation were some of the main motivations behind
> introducing the adaptive scheduler. In the adaptive scheduler we can
> further do things like only to trigger a rescaling operations exactly when
> a checkpoint was completed to minimize reprocessing. For Jobs with small
> state size, the downtime during rescaling can already be << 1 second today.
>
> Chesnay and David Moravek are currently in the process of drafting two
> FLIPs that will extend the support of the adaptive scheduler to session
> mode and will allow clients to change the desired/min/max parallelism of
> the vertices of a Job during its runtime via the REST API. We currently
> plan to publish a draft of these FLIPs next week for discussion. Would you
> consider moving to the adaptive scheduler for the kubernetes operator
> provided these FLIPs make it? I think, it has the potential to simplify the
> logic required for rescaling on the operator side quite a bit.
>
> Best,
>
> Konstantin
>
>
> Am Do., 26. Jan. 2023 um 12:16 Uhr schrieb Maximilian Michels <
> mxm@apache.org>:
>
>> Hey ConradJam,
>>
>> Thank you for your thoughtful response. It would be great to start writing
>> a FLIP for the Rescale API. If you want to take a stab, please go ahead,
>> I'd be happy to review. I'm sure Gyula or others will also chime in.
>>
>> I want to answer your question so we are aligned:
>>
>> ● Does scaling work on YARN, or just k8s?
>> >
>>
>> I think it should work for both YARN and K8s. We would have to make
>> changes
>> to the drivers (AbstractResourceManagerDriver) which is implemented for
>> both K8s and YARN. The outlined approach for rescaling does not require
>> integrating with those systems, just maybe updating how the driver is
>> used,
>> so we should be able to make it work across both YARN and K8s.
>>
>> ● Rescaling supports Standalone mode?
>> >
>>
>> Yes, I think it should and easily can. We do use a different type of
>> resource manager (StandaloneResourceManager, not ActiveResourceManager)
>> but
>> I think the logic will sit on a higher level where the ResourceManager
>> implementation is not relevant.
>>
>> ● Can we simplify the recovery steps?
>> >
>>
>> For the first version, I would prefer the simple approach of (1) acquiring
>> the required slots for rescaling, then (2) trigger a stop with savepoint
>> (3) resubmit the job with updated parallelisms. What you have in mind is a
>> bit more involved but certainly a great optimization, especially when only
>> a fraction of the job state needs to be repartitioned.
>>
>> Of course, there are many details, such as
>> > ● At some point we may not be able to use this kind of hot update, and
>> > still need to restart the job, when this happens, we should prevent
>> users
>> > from using rescaling requests
>> >
>>
>> I'm curious to learn more about "hot updates". How would we support this
>> in
>> Flink? Would we have to support dynamically repartitioning tasks? I don't
>> think Flink supports this yet. For now, restarting the job may be the best
>> we can do.
>>
>> ● After rescaling is submitted, when we fail, there should be a rollback
>> > mechanism to roll back to the previous degree of parallelism.
>> >
>>
>> This should not be necessary if all the requirements for rescaling, e.g.
>> enough task slots, are satisfied by the Rescale API. I'm not even sure
>> rolling back is an option because we can't guarantee that a rollback would
>> always work.
>>
>> Thanks,
>> Max
>>
>> On Tue, Jan 24, 2023 at 6:34 AM ConradJam <ja...@gmail.com> wrote:
>>
>> > Hello max
>> >
>> > Thanks for driving it, I think there is no problem with your previous
>> > suggestion of [1] FLINK-30773. Here I just put forward some supplements
>> and
>> > doubts.I have some suggestions and insights for this
>> >
>> >  I have experienced the autoscaling of Flink K8S Operator for a part of
>> the
>> > time. The current method is to stop the job and modify the parallelism,
>> > which will interrupt the business for a long time. I think the purpose
>> of
>> > modifying Rescaling Api is to better fit cloud native and reduce the
>> impact
>> > of job scaling downtime.
>> >
>> > I have tried scaling with less time, and I call this step "hot update
>> > parallelism" (if there is an available Slots, there is no need to
>> re-deploy
>> > the JobManager Or TaskManager on K8S)
>> >
>> > Around this topic, I raised the *following questions*:
>> > ● Does scaling work on YARN, or just k8s?
>> >    ○ I think we can support running on K8S for the first version, and
>> Yarn
>> > can be considered later
>> > ● Rescaling supports Standalone mode?
>> >    ○ I think it can be supported. The essence is just to modify the
>> > parallelism of job vertices. As for the tuning strategy, it should be
>> > determined by the external system or K8S Operator
>> > ● Can we simplify the recovery steps?
>> >    ○ As far as I know, the traditional way to adjust the parallelism is
>> to
>> > stop a job and do a Savepoint, and then run the job with the adjusted
>> > parallelism. If we hide this step in the *JobManager*, it will be an
>> > important means to reduce the delay.
>> >
>> >   Of course, there are many details, such as
>> > ● At some point we may not be able to use this kind of hot update, and
>> > still need to restart the job, when this happens, we should prevent
>> users
>> > from using rescaling requests
>> > ● After rescaling is submitted, when we fail, there should be a rollback
>> > mechanism to roll back to the previous degree of parallelism.
>> >
>> > more and more ～
>> >
>> >   By the way, because the content may be more, I did not expand more
>> ideas
>> > and descriptions here. This proposal modifies the original Rescaling
>> API.
>> > I would also like to hear if  *@gyula* has some new ideas on this as it
>> was
>> > also involved in the development of FLIP-271
>> > I am willing to write a FLIP for this purpose to improve and write some
>> > ideas with dev Community and then submit it. What do you think about
>> > starting a discussion for the community?
>> >
>> >
>> >    1. https://issues.apache.org/jira/browse/FLINK-30773
>> >
>> > Best～
>> >
>> > Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
>> >
>> > > Hi,
>> > >
>> > > The current rescale API appears to be a work in progress. A couple
>> years
>> > > ago, we disabled access to the API [1].
>> > >
>> > > I'm looking into this problem as part of working on autoscaling [2]
>> where
>> > > we currently require a full restart of the job to apply the
>> parallelism
>> > > overrides. This adds additional delay and comes with the caveat that
>> we
>> > > don't know whether sufficient resources are available prior to
>> executing
>> > > the scaling decision. We obviously do not want to get stuck due to a
>> lack
>> > > of resources. So a rescale API would have to ensure enough resources
>> are
>> > > available prior to restarting the job.
>> > >
>> > > I've created an issue here:
>> > > https://issues.apache.org/jira/browse/FLINK-30773
>> > >
>> > > Any comments or interest in working on this?
>> > >
>> > > -Max
>> > >
>> > > [1] https://issues.apache.org/jira/browse/FLINK-12312
>> > > [2]
>> > >
>> >
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
>> > >
>> >
>> >
>> > --
>> > Best
>> >
>> > ConradJam
>> >
>> >
>> > --
>> > Best
>> >
>> > ConradJam
>> >
>>
>
>
> --
> https://twitter.com/snntrable
> https://github.com/knaufk
>

Re: Reworking the Rescale API

Posted by Konstantin Knauf <kn...@apache.org>.

Hi Max,

it seems to me we are now running in some of the potential duplication of
efforts across the standard and adaptive scheduler that Chesnay had
mentioned on the original ticket. The issue of having to do a full restart
of the Job for rescaling as well as waiting for resources to be available
before doing a rescaling operation were some of the main motivations behind
introducing the adaptive scheduler. In the adaptive scheduler we can
further do things like only to trigger a rescaling operations exactly when
a checkpoint was completed to minimize reprocessing. For Jobs with small
state size, the downtime during rescaling can already be << 1 second today.

Chesnay and David Moravek are currently in the process of drafting two
FLIPs that will extend the support of the adaptive scheduler to session
mode and will allow clients to change the desired/min/max parallelism of
the vertices of a Job during its runtime via the REST API. We currently
plan to publish a draft of these FLIPs next week for discussion. Would you
consider moving to the adaptive scheduler for the kubernetes operator
provided these FLIPs make it? I think, it has the potential to simplify the
logic required for rescaling on the operator side quite a bit.

Best,

Konstantin


Am Do., 26. Jan. 2023 um 12:16 Uhr schrieb Maximilian Michels <
mxm@apache.org>:

> Hey ConradJam,
>
> Thank you for your thoughtful response. It would be great to start writing
> a FLIP for the Rescale API. If you want to take a stab, please go ahead,
> I'd be happy to review. I'm sure Gyula or others will also chime in.
>
> I want to answer your question so we are aligned:
>
> ● Does scaling work on YARN, or just k8s?
> >
>
> I think it should work for both YARN and K8s. We would have to make changes
> to the drivers (AbstractResourceManagerDriver) which is implemented for
> both K8s and YARN. The outlined approach for rescaling does not require
> integrating with those systems, just maybe updating how the driver is used,
> so we should be able to make it work across both YARN and K8s.
>
> ● Rescaling supports Standalone mode?
> >
>
> Yes, I think it should and easily can. We do use a different type of
> resource manager (StandaloneResourceManager, not ActiveResourceManager) but
> I think the logic will sit on a higher level where the ResourceManager
> implementation is not relevant.
>
> ● Can we simplify the recovery steps?
> >
>
> For the first version, I would prefer the simple approach of (1) acquiring
> the required slots for rescaling, then (2) trigger a stop with savepoint
> (3) resubmit the job with updated parallelisms. What you have in mind is a
> bit more involved but certainly a great optimization, especially when only
> a fraction of the job state needs to be repartitioned.
>
> Of course, there are many details, such as
> > ● At some point we may not be able to use this kind of hot update, and
> > still need to restart the job, when this happens, we should prevent users
> > from using rescaling requests
> >
>
> I'm curious to learn more about "hot updates". How would we support this in
> Flink? Would we have to support dynamically repartitioning tasks? I don't
> think Flink supports this yet. For now, restarting the job may be the best
> we can do.
>
> ● After rescaling is submitted, when we fail, there should be a rollback
> > mechanism to roll back to the previous degree of parallelism.
> >
>
> This should not be necessary if all the requirements for rescaling, e.g.
> enough task slots, are satisfied by the Rescale API. I'm not even sure
> rolling back is an option because we can't guarantee that a rollback would
> always work.
>
> Thanks,
> Max
>
> On Tue, Jan 24, 2023 at 6:34 AM ConradJam <ja...@gmail.com> wrote:
>
> > Hello max
> >
> > Thanks for driving it, I think there is no problem with your previous
> > suggestion of [1] FLINK-30773. Here I just put forward some supplements
> and
> > doubts.I have some suggestions and insights for this
> >
> >  I have experienced the autoscaling of Flink K8S Operator for a part of
> the
> > time. The current method is to stop the job and modify the parallelism,
> > which will interrupt the business for a long time. I think the purpose of
> > modifying Rescaling Api is to better fit cloud native and reduce the
> impact
> > of job scaling downtime.
> >
> > I have tried scaling with less time, and I call this step "hot update
> > parallelism" (if there is an available Slots, there is no need to
> re-deploy
> > the JobManager Or TaskManager on K8S)
> >
> > Around this topic, I raised the *following questions*:
> > ● Does scaling work on YARN, or just k8s?
> >    ○ I think we can support running on K8S for the first version, and
> Yarn
> > can be considered later
> > ● Rescaling supports Standalone mode?
> >    ○ I think it can be supported. The essence is just to modify the
> > parallelism of job vertices. As for the tuning strategy, it should be
> > determined by the external system or K8S Operator
> > ● Can we simplify the recovery steps?
> >    ○ As far as I know, the traditional way to adjust the parallelism is
> to
> > stop a job and do a Savepoint, and then run the job with the adjusted
> > parallelism. If we hide this step in the *JobManager*, it will be an
> > important means to reduce the delay.
> >
> >   Of course, there are many details, such as
> > ● At some point we may not be able to use this kind of hot update, and
> > still need to restart the job, when this happens, we should prevent users
> > from using rescaling requests
> > ● After rescaling is submitted, when we fail, there should be a rollback
> > mechanism to roll back to the previous degree of parallelism.
> >
> > more and more ～
> >
> >   By the way, because the content may be more, I did not expand more
> ideas
> > and descriptions here. This proposal modifies the original Rescaling API.
> > I would also like to hear if  *@gyula* has some new ideas on this as it
> was
> > also involved in the development of FLIP-271
> > I am willing to write a FLIP for this purpose to improve and write some
> > ideas with dev Community and then submit it. What do you think about
> > starting a discussion for the community?
> >
> >
> >    1. https://issues.apache.org/jira/browse/FLINK-30773
> >
> > Best～
> >
> > Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
> >
> > > Hi,
> > >
> > > The current rescale API appears to be a work in progress. A couple
> years
> > > ago, we disabled access to the API [1].
> > >
> > > I'm looking into this problem as part of working on autoscaling [2]
> where
> > > we currently require a full restart of the job to apply the parallelism
> > > overrides. This adds additional delay and comes with the caveat that we
> > > don't know whether sufficient resources are available prior to
> executing
> > > the scaling decision. We obviously do not want to get stuck due to a
> lack
> > > of resources. So a rescale API would have to ensure enough resources
> are
> > > available prior to restarting the job.
> > >
> > > I've created an issue here:
> > > https://issues.apache.org/jira/browse/FLINK-30773
> > >
> > > Any comments or interest in working on this?
> > >
> > > -Max
> > >
> > > [1] https://issues.apache.org/jira/browse/FLINK-12312
> > > [2]
> > >
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
> > >
> >
> >
> > --
> > Best
> >
> > ConradJam
> >
> >
> > --
> > Best
> >
> > ConradJam
> >
>


-- 
https://twitter.com/snntrable
https://github.com/knaufk

Re: Reworking the Rescale API

Posted by Maximilian Michels <mx...@apache.org>.

Hey ConradJam,

Thank you for your thoughtful response. It would be great to start writing
a FLIP for the Rescale API. If you want to take a stab, please go ahead,
I'd be happy to review. I'm sure Gyula or others will also chime in.

I want to answer your question so we are aligned:

● Does scaling work on YARN, or just k8s?
>

I think it should work for both YARN and K8s. We would have to make changes
to the drivers (AbstractResourceManagerDriver) which is implemented for
both K8s and YARN. The outlined approach for rescaling does not require
integrating with those systems, just maybe updating how the driver is used,
so we should be able to make it work across both YARN and K8s.

● Rescaling supports Standalone mode?
>

Yes, I think it should and easily can. We do use a different type of
resource manager (StandaloneResourceManager, not ActiveResourceManager) but
I think the logic will sit on a higher level where the ResourceManager
implementation is not relevant.

● Can we simplify the recovery steps?
>

For the first version, I would prefer the simple approach of (1) acquiring
the required slots for rescaling, then (2) trigger a stop with savepoint
(3) resubmit the job with updated parallelisms. What you have in mind is a
bit more involved but certainly a great optimization, especially when only
a fraction of the job state needs to be repartitioned.

Of course, there are many details, such as
> ● At some point we may not be able to use this kind of hot update, and
> still need to restart the job, when this happens, we should prevent users
> from using rescaling requests
>

I'm curious to learn more about "hot updates". How would we support this in
Flink? Would we have to support dynamically repartitioning tasks? I don't
think Flink supports this yet. For now, restarting the job may be the best
we can do.

● After rescaling is submitted, when we fail, there should be a rollback
> mechanism to roll back to the previous degree of parallelism.
>

This should not be necessary if all the requirements for rescaling, e.g.
enough task slots, are satisfied by the Rescale API. I'm not even sure
rolling back is an option because we can't guarantee that a rollback would
always work.

Thanks,
Max

On Tue, Jan 24, 2023 at 6:34 AM ConradJam <ja...@gmail.com> wrote:

> Hello max
>
> Thanks for driving it, I think there is no problem with your previous
> suggestion of [1] FLINK-30773. Here I just put forward some supplements and
> doubts.I have some suggestions and insights for this
>
>  I have experienced the autoscaling of Flink K8S Operator for a part of the
> time. The current method is to stop the job and modify the parallelism,
> which will interrupt the business for a long time. I think the purpose of
> modifying Rescaling Api is to better fit cloud native and reduce the impact
> of job scaling downtime.
>
> I have tried scaling with less time, and I call this step "hot update
> parallelism" (if there is an available Slots, there is no need to re-deploy
> the JobManager Or TaskManager on K8S)
>
> Around this topic, I raised the *following questions*:
> ● Does scaling work on YARN, or just k8s?
>    ○ I think we can support running on K8S for the first version, and Yarn
> can be considered later
> ● Rescaling supports Standalone mode?
>    ○ I think it can be supported. The essence is just to modify the
> parallelism of job vertices. As for the tuning strategy, it should be
> determined by the external system or K8S Operator
> ● Can we simplify the recovery steps?
>    ○ As far as I know, the traditional way to adjust the parallelism is to
> stop a job and do a Savepoint, and then run the job with the adjusted
> parallelism. If we hide this step in the *JobManager*, it will be an
> important means to reduce the delay.
>
>   Of course, there are many details, such as
> ● At some point we may not be able to use this kind of hot update, and
> still need to restart the job, when this happens, we should prevent users
> from using rescaling requests
> ● After rescaling is submitted, when we fail, there should be a rollback
> mechanism to roll back to the previous degree of parallelism.
>
> more and more ～
>
>   By the way, because the content may be more, I did not expand more ideas
> and descriptions here. This proposal modifies the original Rescaling API.
> I would also like to hear if  *@gyula* has some new ideas on this as it was
> also involved in the development of FLIP-271
> I am willing to write a FLIP for this purpose to improve and write some
> ideas with dev Community and then submit it. What do you think about
> starting a discussion for the community?
>
>
>    1. https://issues.apache.org/jira/browse/FLINK-30773
>
> Best～
>
> Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：
>
> > Hi,
> >
> > The current rescale API appears to be a work in progress. A couple years
> > ago, we disabled access to the API [1].
> >
> > I'm looking into this problem as part of working on autoscaling [2] where
> > we currently require a full restart of the job to apply the parallelism
> > overrides. This adds additional delay and comes with the caveat that we
> > don't know whether sufficient resources are available prior to executing
> > the scaling decision. We obviously do not want to get stuck due to a lack
> > of resources. So a rescale API would have to ensure enough resources are
> > available prior to restarting the job.
> >
> > I've created an issue here:
> > https://issues.apache.org/jira/browse/FLINK-30773
> >
> > Any comments or interest in working on this?
> >
> > -Max
> >
> > [1] https://issues.apache.org/jira/browse/FLINK-12312
> > [2]
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
> >
>
>
> --
> Best
>
> ConradJam
>
>
> --
> Best
>
> ConradJam
>

Re: Reworking the Rescale API

Posted by ConradJam <ja...@gmail.com>.

Hello max

Thanks for driving it, I think there is no problem with your previous
suggestion of [1] FLINK-30773. Here I just put forward some supplements and
doubts.I have some suggestions and insights for this

 I have experienced the autoscaling of Flink K8S Operator for a part of the
time. The current method is to stop the job and modify the parallelism,
which will interrupt the business for a long time. I think the purpose of
modifying Rescaling Api is to better fit cloud native and reduce the impact
of job scaling downtime.

I have tried scaling with less time, and I call this step "hot update
parallelism" (if there is an available Slots, there is no need to re-deploy
the JobManager Or TaskManager on K8S)

Around this topic, I raised the *following questions*:
● Does scaling work on YARN, or just k8s?
   ○ I think we can support running on K8S for the first version, and Yarn
can be considered later
● Rescaling supports Standalone mode?
   ○ I think it can be supported. The essence is just to modify the
parallelism of job vertices. As for the tuning strategy, it should be
determined by the external system or K8S Operator
● Can we simplify the recovery steps?
   ○ As far as I know, the traditional way to adjust the parallelism is to
stop a job and do a Savepoint, and then run the job with the adjusted
parallelism. If we hide this step in the *JobManager*, it will be an
important means to reduce the delay.

  Of course, there are many details, such as
● At some point we may not be able to use this kind of hot update, and
still need to restart the job, when this happens, we should prevent users
from using rescaling requests
● After rescaling is submitted, when we fail, there should be a rollback
mechanism to roll back to the previous degree of parallelism.

more and more ～

  By the way, because the content may be more, I did not expand more ideas
and descriptions here. This proposal modifies the original Rescaling API.
I would also like to hear if  *@gyula* has some new ideas on this as it was
also involved in the development of FLIP-271
I am willing to write a FLIP for this purpose to improve and write some
ideas with dev Community and then submit it. What do you think about
starting a discussion for the community?


   1. https://issues.apache.org/jira/browse/FLINK-30773

Best～

Maximilian Michels <mx...@apache.org> 于2023年1月24日周二 01:08写道：

> Hi,
>
> The current rescale API appears to be a work in progress. A couple years
> ago, we disabled access to the API [1].
>
> I'm looking into this problem as part of working on autoscaling [2] where
> we currently require a full restart of the job to apply the parallelism
> overrides. This adds additional delay and comes with the caveat that we
> don't know whether sufficient resources are available prior to executing
> the scaling decision. We obviously do not want to get stuck due to a lack
> of resources. So a rescale API would have to ensure enough resources are
> available prior to restarting the job.
>
> I've created an issue here:
> https://issues.apache.org/jira/browse/FLINK-30773
>
> Any comments or interest in working on this?
>
> -Max
>
> [1] https://issues.apache.org/jira/browse/FLINK-12312
> [2]
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-271%3A+Autoscaling
>


-- 
Best

ConradJam


-- 
Best

ConradJam