Thanks Till for the explanation and the follow up actions.
That sounds good to me.
Thanks,
Zhu
Till Rohrmann <trohrm...@apache.org> 于2021年1月26日周二 下午7:28写道：
Thanks a lot for all your feedback. Let me try to answer it.
# ScaleUpController vs. RescaleController
At the moment, the idea of the declarative scheduler is to run a job with
a parallelism which is as close to the target value as possible but to
never exceed it. Since the target value is currently fixed (this will
change once auto-scaling is supported), we never have to scale down. In
fact, scale down events only happen if the system loses slots and then
you
cannot choose whether to scale down or not. This and the idea to keep the
initial design as simple as possible motivated the naming of
ScaleUpController. Once we support auto-scaling, we might have to rename
this interface. However, since I don't fully know how things will look
like
with auto-scaling, I would like to refrain from this change now.
# Relationship between declarative scheduler and existing implementations
The declarative scheduler will implement the SchedulerNG interface. At
the
moment I am not aware of any required changes to this interface. Also the
way the JobMaster interacts with the scheduler won't change to the best
of
my knowledge.
# Awareness of SlotAllocator of the ExecutionGraph
The idea is to make the SlotAllocator not aware of the ExecutionGraph
because it can only be created after the SlotAllocator has decided on the
parallelism the job can be run with. Moreover, the idea is that the
JobInformation instance contains all the required information of a job to
make this decision.
The SlotAllocator also reserves the slots for a job before the
ExecutionGraph is created. Consequently, we need a way to associate the
slots with the Executions of the ExecutionGraph. Here we have decided to
use the ExecutionVertexID as the identificator. We could also introduce a
new identificator as long as we can map this one to the
ExecutionVertexID.
We could for example use JobVertexID and subtask id as the identificator
but this is exactly what the ExecutionVertexID is. That's why we decided
to
reuse it for the time being.
# ScaleUpController
## Cumulative parallelism
Yes, the cumulative parallelism is the sum of all tasks.
## When to call the ScaleUpController
Yes, the idea is that the scheduler will check whether one can run the
job
with an increased parallelism if there are new slots available. If this
is
the case, then we will ask the ScaleUpController, whether we actually
should scale up (e.g. whether the increase is significant enough or
whether
enough time has passed between the last scale up operation).
## Do we provide enough information to the ScaleUpController
I am pretty sure that we don't provide the ScaleUpController enough
information to make the best possible decision. We wanted to keep it
simple
in the first version and iterate on the interface with the help of user
feedback. I think that this interface won't fundamentally change the
scheduler and, hence, it shouldn't block future extensions by starting
with
a simple interface.
# Cluster vs. job configuration
I think you are right that it would be most flexible if one could select
a
scheduler on a per job basis with falling back to the cluster
configuration
if nothing has been specified. For the sake of simplicity and narrowing
down the scope I would consider this as a follow up.
# Performance on failover
I agree that the performance won't be optimal in the failover case
because
of 1) having to recreate the EG and 2) redundant IO operations. For 1) I
agree that FLINK-21110 will help a lot. For 2) moving the IO operations
out
of the EG could be a good improvement. With the same argument as above, I
would consider this a follow up because I would like to keep the initial
design as simple as possible and I think that these performance
optimizations are not required to make this feature work.
# Lost execution history
You are right Zhu Zhu that recreating the ExecutionGraph causes the loss
of information. We are currently investigating which information is
strictly required and needs to be maintained across ExecutionGraph
creations (mainly for tests to succeed). One idea is to either store this
information outside of the ExecutionGraph or to initialize the
ExecutionGraph with the information from the previous instance. Most
likely, we won't give a guarantee that all counters, timestamps and
metrics
are correctly maintained across failovers in the first version, though.
It
is a bit the same argument as above that this is not strictly required to
make this feature work. Maybe this means that this feature is not
production ready for some users, but I think this is ok.
In general, to fully integrate the declarative scheduler with the web ui
we have to be able to display changing ExecutionGraphs. Ideally, we would
have something like a timeline where one can select the time for which to
display the state of the job execution. If one goes all the way to the
right, the system shows the live state and all other positions are the
present time minus some offset.
I will add the follow ups to the FLIP as potential improvements in order
to keep track of them.
Cheers,
Till
On Tue, Jan 26, 2021 at 9:53 AM Zhu Zhu <reed...@gmail.com> wrote:
Thanks for the proposal! @Till Rohrmann <trohrm...@apache.org>
The design looks generally good to me. I have 2 concerns though:
# performance on failover
I can see is that an ExecutionGraph will be built and initialized on
each
task failure. The process is possibly to be slow:
1. the topology building can be very slow (much slower than the
scheduling
or restarting process) when the job scale becomes large (see
FLINK-20612).
Luckily FLINK-21110 will improve it.
2. the input/output format can be slow due to possible IO work to
external services.
Maybe we should extract it out from ExecutionGraph building?
# execution history lost
After building and using a new ExecutionGraph, the execution history in
the
old graph will be lost, including status timestamps of the job, prior
execution
attempts and their failures. Should we let the new EG inherit these
information
from prior EG? Maybe as a future plan with further discussions regarding
the
varying parallelism.
Thanks,
Zhu
Xintong Song <tonysong...@gmail.com> 于2021年1月24日周日 上午11:55写道：
Thanks for preparing the FLIP and starting the discussion, Till.
I have a few questions trying to understand the design.
## What is the relationship between the current and new schedulers?
IIUC, the new declarative scheduler will coexist with the current
scheduler, as an alternative that the user needs to explicitly switch
to.
Then does it require any changes to the scheduler interfaces and how
the
JobMaster interacts with it?
## Is `SlotAllocator` aware of `ExecutionGraph`?
Looking at the interfaces, it seems to me that `SlotAllocator` only
takes
`JobInformation` and `VertexInformation` as topology inputs. However,
it
generates `ParallelismAndResourceAssignments` which maps slots to
`ExecutionVertexID`s. It's a bit confusing that `ExecutionGraph` is
generated outside `SlotAllocator` while `ExecutionVertexID`s are
generated
inside `SlotAllocator`.
## About `ScaleUpController#canScaleUp`
### What is cumulative parallelism?
The interface shows that the cumulative parallelism is a single number
per
job graph. I assume this is the sum of parallelism of all vertices?
### Is this interface expected to be called before or after
`SlotAllocator#determineParallelism`?
IIUC, on new resources appear, the scheduler always calls
`SlotAllocator#determineParallelism` to generate a new plan based on
the
available resources, and `ScaleUpController#canScaleUp` is then called
to
decide whether to apply the new plan, according to whether the
increasement
is significant, how long the job has been running since last restart,
etc.
Is that correct?
### The cumulative parallelism may not be enough for deciding whether
the
job should be scaled up.
I'm assuming my understanding for the above questions are correct.
Please
correct me if not.
I noticed Chesnay's comment on the wiki page about making the decision
based on when the job entered the execution state last time.
In addition to that, there could also be situations where jobs may not
benefit much an increment in cumulative parallelism.
E.g., for a topology A -> B, where A and B are in different slot
sharing
groups, and the current parallelism for both A and B are 2. When 1 new
slot
appears, `SlotAllocator` may suggest increasing parallelism of A to 3.
But
this may not be a significant beneficial change for the job because the
overall performance is still bottlenecked by the parallelism of B.
## Cluster vs. Job configuration
I'm not entirely sure about specifying which scheduler to be used via a
cluster level configuration option. Each job in a Flink cluster has its
own
JobMaster, and which scheduler to use is internal to that JobMaster. I
understand that the declarative scheduler requires the cluster to use
declarative resource management. However, other jobs should still be
allowed to use other scheduler implementations that also support
declarative resource management (e.g. the current `DefaultScheduler`).
Maybe we should consider the cluster level configuration option as a
default scheduler, and allow the job to specify a different scheduler
in
its execution config. This is similar to how we specify which state
backend
to be used.
## Minor: It might be better to also show in the state machine figure
that
it can go from `Executing` to `Restarting` when new resources appear.
Thank you~
Xintong Song
On Sat, Jan 23, 2021 at 6:04 AM Steven Wu <stevenz...@gmail.com>
wrote:
Till, thanks a lot for the proposal.
Even if the initial phase is only to support scale-up, maybe the
"ScaleUpController" interface should be called "RescaleController" so
that
in the future scale-down can be added.
On Fri, Jan 22, 2021 at 7:03 AM Till Rohrmann <trohrm...@apache.org>
wrote:
Hi everyone,
I would like to start a discussion about adding a new type of
scheduler
to
Flink. The declarative scheduler will first declare the required
resources
and wait for them before deciding on the actual parallelism of a
job.
Thereby it can better handle situations where resources cannot be
fully
fulfilled. Moreover, it will act as a building block for the
reactive
mode
where Flink should scale to the maximum of the currently available
resources.
Please find more details in the FLIP wiki document [1]. Looking
forward
to
your feedback.
[1] https://cwiki.apache.org/confluence/x/mwtRCg
Cheers,
Till
