Thanks Chesnay&Till for proposing this improvement.
It's of good value to allow jobs to make best use of available
resources adaptively. Not
to mention it further supports reactive mode.
So big +1 for it.
I have a minor concern about possible regression in certain cases due
to the proposed
JobVertex-wise scheduling which replaces current ExecutionVertex-wise
scheduling.
In the proposal, looks to me it requires a stage to finish before its
consumer stage can be
scheduled. This limitation, however, does not exist in current
scheduler. In the case that there
exists a POINTWISE BLOCKING edge, the downstream execution region can
be scheduled
right after its connected upstream execution vertices finishes, even
before the whole upstream
stage finishes. This allows the region to be launched earlier and make
use of available resources.
Do we need to let the new scheduler retain this property?
Thanks,
Zhu
Xintong Song <tonysong...@gmail.com <mailto:tonysong...@gmail.com>>
于2020年8月26日周三 下午6:59写道：
Thanks for the quick response.
*Job prioritization, Allocation IDs, Minimum resource
requirements, SlotManager Implementation Plan:* Sounds good to me.
*FLIP-56*
Good point about the trade-off. I believe maximum resource
utilization and
quick deployment are desired in different scenarios. E.g., a long
running
streaming job deserves some deployment latency to improve the resource
utilization, which benefits the entire lifecycle of the job. On
the other
hand, short batch queries may prefer quick deployment, otherwise
the time
for resource allocation might significantly increase the response
time.
It would be good enough for me to bring these questions to attention.
Nothing that I'm aware of should block this FLIP.
Thank you~
Xintong Song
On Wed, Aug 26, 2020 at 5:14 PM Chesnay Schepler
<ches...@apache.org <mailto:ches...@apache.org>> wrote:
> Thank you Xintong for your questions!
> Job prioritization
> Yes, the job which declares it's initial requirements first is
prioritized.
> This is very much for simplicity; for example this avoids the
nasty case
> where all jobs get some resources, but none get enough to
actually run the
> job.
> Minimum resource requirements
>
> My bad; at some point we want to allow the JobMaster to declare
a range of
> resources it could use to run a job, for example min=1, target=10,
> max=+inf.
>
> With this model, the RM would then try to balance the resources
such that
> as many jobs as possible are as close to the target state as
possible.
>
> Currently, the minimum/target/maximum resources are all the
same. So the
> notification is sent whenever the current requirements cannot be
met.
> Allocation IDs
> We do intend to, at the very least, remove AllocationIDs on the
> SlotManager side, as they are just not required there.
>
> On the slotpool side we have to keep them around at least until the
> existing Slotpool implementations are removed (not sure whether
we'll fully
> commit to this in 1.12), since the interfaces use AllocationIDs,
which also
> bleed into the JobMaster.
> The TaskExecutor is in a similar position.
> But in the long-term, yes they will be removed, and most usages will
> probably be replaced by the SlotID.
> FLIP-56
>
> Dynamic slot allocations are indeed quite interesting and raise
a few
> questions; for example, the main purpose of it is to ensure maximum
> resource utilization. In that case, should the JobMaster be
allowed to
> re-use a slot it if the task requires less resources than the slot
> provides, or should it always request a new slot that exactly
matches?
>
> There is a trade-off to be made between maximum resource utilization
> (request exactly matching slots, and only re-use exact matches)
and quicker
> job deployment (re-use slot even if they don't exactly match, skip
> round-trip to RM).
>
> As for how to handle the lack of a preemptively known SlotIDs,
that should
> be fine in and of itself; we already handle a similar case when
we request
> a new TaskExecutor to be started. So long as there is some way
to know how
> many resources the TaskExecutor has in total I do not see a
problem at the
> moment. We will get the SlotID eventually by virtue of the heartbeat
> SlotReport.
> Implementation plan (SlotManager)
> You are on the right track. The SlotManager tracks the declared
resource
> requirements, and if the requirements increased it creates a
SlotRequest,
> which then goes through similar code paths as we have at the
moment (try to
> find a free slot, if found tell the TM, otherwise try to request
new TM).
> The SlotManager changes are not that substantial to get a
working version;
> we have a PoC and most of the work went into refactoring the
SlotManager
> into a more manageable state. (split into several components,
stricter and
> simplified Slot life-cycle, ...).
> Offer/free slots between JM/TM
> Gotta run, but that's a good question and I'll think about. But
I think it
> comes down to making less changes, and being able to leverage
existing
> reconciliation protocols.
> Do note that TaskExecutor also explicitly inform the RM about
freed slots;
> the heartbeat slot report is just a safety net.
> I'm not sure whether slot requests are able to overtake a slot
release;
> @till do you have thoughts on that?
> As for the race condition between the requirements reduction and
slot
> release, if we run into problems we have the backup plan of only
releasing
> the slot after the requirement reduction has been acknowledged.
>
> On 26/08/2020 10:31, Xintong Song wrote:
>
> Thanks for preparing the FLIP and driving this discussion,
@Chesnay & @Till.
>
> I really like the idea. I see a great value in the proposed
declarative
> resource management, in terms of flexibility, usability and
efficiency.
>
> I have a few comments and questions regarding the FLIP design.
In general,
> the protocol design makes good sense to me. My main concern is
that it is
> not very clear to me what changes are required from the
> Resource/SlotManager side to adapt to the new protocol.
>
> *1. Distributed slots across different jobs*
>
> Jobs which register their requirements first, will have
precedence over
>
> other jobs also if the requirements change during the runtime.
>
> Just trying to understand, does this mean jobs are prioritized
by the order
> of their first resource declaring?
>
> *2. AllocationID*
>
> Is this FLIP suggesting to completely remove AllocationID?
>
> I'm fine with this change. It seems where AllocationID is used
can either
> be removed or be replaced by JobID. This reflects the concept
that slots
> are now assigned to a job instead of its individual slot requests.
>
> I would like to bring to attention that this also requires
changes on the
> TM side, with respect to FLIP-56[1].
>
> In the context of dynamic slot allocation introduced by FLIP-56,
slots do
> not pre-exist on TM and are dynamically created when RM calls
> TaskExecutorGateway.requestSlot. Since the slots do not
pre-exist, nor
> their SlotIDs, RM requests slots from TM with a special SlotID
(negative
> slot index). The semantic changes from "requesting the slot
identified by
> the given SlotID" to "requesting a slot with the given resource
profile".
> The AllocationID is used for identifying the dynamic slots in
such cases.
>
> >From the perspective of FLIP-56 and fine grained resource
management, I'm
> fine with removing AllocationID. In the meantime, we would need
TM to
> recognize the special negative indexed SlotID and generate a new
unique
> SlotID for identifying the slot.
>
> *3. Minimum resource requirement*
>
> However, we can let the JobMaster know if we cannot fulfill the
minimum
>
> resource requirement for a job after
> resourcemanager.standalone.start-up-time has passed.
>
> What is the "minimum resource requirement for a job"? Did I overlook
> anything?
>
> *4. Offer/free slots between JM/TM*
>
> This probably deserves a separate discussion thread. Just want
to bring it
> up.
>
> The idea has been coming to me for quite some time. Is this
design, that JM
> requests resources from RM while accepting/releasing resources
from/to TM,
> the right thing?
>
> The pain point is that events of JM's activities
(requesting/releasing
> resources) arrive at RM out of order. This leads to several
problems.
>
> - When a job fails and task cancelation takes long, some of
the slots
> might be released from the slot pool due to being unused for
a while. Then
> the job restarts and requests these slots again. At this
time, RM may
> receive slot requests before noticing from TM heartbeats that
previous
> slots are released, thus requesting new resources. I've seen
many times
> that the Yarn cluster has a heavy load and is not allocating
resources
> quickly enough, which leads to slot request timeout and job
failover, and
> during the failover more resources are requested which adds
more load to
> the Yarn cluster. Happily, this should be improved with the
declarative
> resource management. :)
> - As described in this FLIP, it is possible that RM learns
the releasing
> of slots from TM heartbeat before noticing the resource
requirement
> decreasing, it may allocate more resources which need to be
released soon.
> - It complicates the ResourceManager/SlotManager, by requiring an
> additional slot state PENDING, which means the slot is
assigned by RM but
> is not confirmed successfully ordered by TM.
>
> Why not just make RM offer the allocated resources (TM address,
SlotID,
> etc.) to JM, and JM release resources to RM? So that for all the
resource
> management JM talks to RM, and for the task deployment and
execution it
> talks to TM?
>
> I tried to understand the benefits for having the current
design, and found
> the following in FLIP-6[2].
>
>
> All that the ResourceManager does is negotiate between the
> cluster-manager, the JobManager, and the TaskManagers. Its state
can hence
> be reconstructed from re-acquiring containers and
re-registration from
> JobManagers and TaskManagers
>
> Correct me if I'm wrong, it seems the original purpose is to
make sure the
> assignment between jobs and slots are confirmed between JM and
TMs, so that
> failures of RM will not lead to any inconsistency. However, this
only
> benefits scenarios where RM fails while JM and TMs live.
Currently, JM and
> RM are in the same process. We do not really have any scenario
where RM
> fails alone. We might separate JM and RM to different processes
in future,
> but as far as I can see we don't have such requirements at the
moment. It
> seems to me that we are suffering the current problems, complying to
> potential future benefits.
>
> Maybe I overlooked something.
>
> *5. Implementation Plan*
>
> For SlotPool, it sounds quite straightforward to "aggregate
individual slot
> requests".
>
> For Resource/SlotManager, it seems there are quite a lot changes
needed,
> with the removal of individual slot requests and AllocationID.
It's not
> clear to me what is the first step plan for RM/SM? Do we
internally treat
> the resource requirements as individual slot requests as the
first step, so
> only the interfaces are changed? Or do we actually change
(practically
> re-write) the slot allocation logics?
>
> Thank you~
>
> Xintong Song
>
>
>
[1]https://cwiki.apache.org/confluence/display/FLINK/FLIP-56%3A+Dynamic+Slot+Allocation
>
[2]https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=65147077
>
> On Tue, Aug 25, 2020 at 4:48 PM Chesnay Schepler
<ches...@apache.org <mailto:ches...@apache.org>>
<ches...@apache.org <mailto:ches...@apache.org>> wrote:
>
>
> Hello,
>
> in FLIP-138 we want to rework the way the JobMaster acquires
slots, such
> that required resources are declared before a job is scheduled
and th
> job execution is adjusted according to the provided resources (e.g.,
> reducing parallelism), instead of asking for a fixed number of
resources
> during scheduling and failing midway through if not enough
resources are
> available.
>
> This is a stepping stone towards reactive mode, where Flink will
> automatically make use of new TaskExecutors being started.
>
> More details can be found here
>
<https://cwiki.apache.org/confluence/display/FLINK/FLIP-138%3A+Declarative+Resource+management
>
> .
>
>
>
