Hey Folks,
Thanks for the discussion!
*Motiviation and use cases*
I think motiviation and use cases are very clear and I do not have doubts
on this part.
A typical use case is ETL with two-phase-commit, hundreds of partitions can
be blocked by a single straggler (a single task's checkpoint abortion can
affect all, not necessary failure).
*Source offset redistribution*
As for the known sources & implementation for Flink, I can not find a case
that does not work, *for now*.
I need to dig a bit more: how splits are tracked assigned, not successfully
processed, succesffully processed e.t.c.
I guess it is a single shared source OPCoordinator. And how this *shared*
state (between tasks) is preserved?
*Input partition/splits treated completely independent from each other*
This part I am still not sure, as mentioned if we have shared state of
source in the above section.
To Thomas:
In Yuan's example, is there a reason why CP8 could not be promoted to
CP10 by the coordinator for PR2 once it receives the notification that
CP10 did not complete? It appears that should be possible since in its
effect it should be no different than no data processed between CP8
and CP10?
Not sure what "promoted" means here, but
1. I guess it does not matter whether it is CP8 or CP10 any more,
if no shared state in source, as exactly what you meantinoed,
"it should be no different than no data processed between CP8 and CP10"
2. I've noticed that from this question there is a gap between
"*allow aborted/failed checkpoint in independent sub-graph*" and
my intention: "*independent sub-graph checkpointing indepently*"
Best
Yuan
On Tue, Feb 8, 2022 at 11:34 AM Gen Luo <luogen...@gmail.com> wrote:
Hi,
I'm thinking about Yuan's case. Let's assume that the case is running in
current Flink:
1. CP8 finishes
2. For some reason, PR2 stops consuming records from the source (but is
not
stuck), and PR1 continues consuming new records.
3. CP9 and CP10 finish
4. PR2 starts to consume quickly to catch up with PR1, and reaches the
same
final status with that in Yuan's case before CP11 starts.
I support that in this case, the status of the job can be the same as in
Yuan's case, and the snapshot (including source states) taken at CP10
should be the same as the composed global snapshot in Yuan's case, which
is
combining CP10 of PR1 and CP8 of PR2. This should be true if neither
failed
checkpointing nor uncommitted consuming have side effects, both of which
can break the exactly-once semantics when replaying. So I think there
should be no difference between rescaling the combined global snapshot or
the globally taken one, i.e. if the input partitions are not independent,
we are probably not able to rescale the source state in the current Flink
eiter.
And @Thomas, I do agree that the operational burden is
significantly reduced, while I'm a little afraid that checkpointing the
subgraphs individually may increase most of the runtime overhead back
again. Maybe we can find a better way to implement this.
On Tue, Feb 8, 2022 at 5:11 AM Thomas Weise <t...@apache.org> wrote:
Hi,
Thanks for opening this discussion! The proposed enhancement would be
interesting for use cases in our infrastructure as well.
There are scenarios where it makes sense to have multiple disconnected
subgraphs in a single job because it can significantly reduce the
operational burden as well as the runtime overhead. Since we allow
subgraphs to recover independently, then why not allow them to make
progress independently also, which would imply that checkpointing must
succeed for non affected subgraphs as certain behavior is tied to
checkpoint completion, like Kafka offset commit, file output etc.
As for source offset redistribution, offset/position needs to be tied
to splits (in FLIP-27) and legacy sources. (It applies to both Kafka
and Kinesis legacy sources and FLIP-27 Kafka source.). With the new
source framework, it would be hard to implement a source with correct
behavior that does not track the position along with the split.
In Yuan's example, is there a reason why CP8 could not be promoted to
CP10 by the coordinator for PR2 once it receives the notification that
CP10 did not complete? It appears that should be possible since in its
effect it should be no different than no data processed between CP8
and CP10?
Thanks,
Thomas
On Mon, Feb 7, 2022 at 2:36 AM Till Rohrmann <trohrm...@apache.org>
wrote:
Thanks for the clarification Yuan and Gen,
I agree that the checkpointing of the sources needs to support the
rescaling case, otherwise it does not work. Is there currently a
source
implementation where this wouldn't work? For Kafka it should work
because
we store the offset per assigned partition. For Kinesis it is
probably
the
same. For the Filesource we store the set of unread input splits in
the
source coordinator and the state of the assigned splits in the
sources.
This should probably also work since new splits are only handed out
to
running tasks.
Cheers,
Till
On Mon, Feb 7, 2022 at 10:29 AM Yuan Mei <yuanmei.w...@gmail.com>
wrote:
Hey Till,
Why rescaling is a problem for pipelined regions/independent
execution
subgraphs:
Take a simplified example :
job graph : source (2 instances) -> sink (2 instances)
execution graph:
source (1/2) -> sink (1/2) [pieplined region 1]
source (2/2) -> sink (2/2) [pieplined region 2]
Let's assume checkpoints are still triggered globally, meaning
different
pipelined regions share the global checkpoint id (PR1 CP1 matches
with
PR2
CP1).
Now let's assume PR1 completes CP10 and PR2 completes CP8.
Let's say we want to rescale to parallelism 3 due to increased
input.
- Notice that we can not simply rescale based on the latest
completed
checkpoint (CP8), because PR1 has already had data (CP8 -> CP10)
output
externally.
- Can we take CP10 from PR1 and CP8 from PR2? I think it depends on
how the
source's offset redistribution is implemented.
The answer is yes if we treat each input partition as
independent
from
each other, *but I am not sure whether we can make that
assumption*.
If not, the rescaling cannot happen until PR1 and PR2 are aligned
with
CPs.
Best
-Yuan
On Mon, Feb 7, 2022 at 4:17 PM Till Rohrmann <trohrm...@apache.org
wrote:
Hi everyone,
Yuan and Gen could you elaborate why rescaling is a problem if we
say
that
separate pipelined regions can take checkpoints independently?
Conceptually, I somehow think that a pipelined region that is
failed
and
cannot create a new checkpoint is more or less the same as a
pipelined
region that didn't get new input or a very very slow pipelined
region
which
couldn't read new records since the last checkpoint (assuming
that
the
checkpoint coordinator can create a global checkpoint by
combining
individual checkpoints (e.g. taking the last completed checkpoint
from
each
pipelined region)). If this comparison is correct, then this
would
mean
that we have rescaling problems under the latter two cases.
Cheers,
Till
On Mon, Feb 7, 2022 at 8:55 AM Gen Luo <luogen...@gmail.com>
wrote:
Hi Gyula,
Thanks for sharing the idea. As Yuan mentioned, I think we can
discuss
this
within two scopes. One is the job subgraph, the other is the
execution
subgraph, which I suppose is the same as PipelineRegion.
An idea is to individually checkpoint the PipelineRegions, for
the
recovering in a single run.
Flink has now supported PipelineRegion based failover, with a
subset
of a
global checkpoint snapshot. The checkpoint barriers are spread
within a
PipelineRegion, so the checkpointing of individual
PipelineRegions
is
actually independent. Since in a single run of a job, the
PipelineRegions
are fixed, we can individually checkpoint separated
PipelineRegions,
despite what status the other PipelineRegions are, and use a
snapshot
of
a
failing region to recover, instead of the subset of a global
snapshot.
This
can support separated job subgraphs as well, since they will
also
be
separated into different PipelineRegions. I think this can
fulfill
your
needs.
In fact the individual snapshots of all PipelineRegions can
form
a
global
snapshot, and the alignment of snapshots of individual regions
is
not
necessary. But rescaling this global snapshot can be
potentially
complex. I
think it's better to use the individual snapshots in a single
run,
and
take
a global checkpoint/savepoint before restarting the job,
rescaling
it
or
not.
A major issue of this plan is that it breaks the checkpoint
mechanism
of
Flink. As far as I know, even in the approximate recovery, the
snapshot
used to recover a single task is still a part of a global
snapshot. To
implement the individual checkpointing of PipelineRegions,
there
may
need
to be a checkpoint coordinator for each PipelineRegion, and a
new
global
checkpoint coordinator. When the scale goes up, there can be
many
individual regions, which can be a big burden to the job
manager.
The
meaning of the checkpoint id will also be changed, which can
affect
many
aspects. There can be lots of work and risks, and the risks
still
exist
if
we only individually checkpoint separated job subgraphs, since
the
mechanism is still broken. If that is what you need, maybe
separating
them
into different jobs is an easier and better choice, as Caizhi
and
Yuan
mentioned.
On Mon, Feb 7, 2022 at 11:39 AM Yuan Mei <
yuanmei.w...@gmail.com
wrote:
Hey Gyula,
That's a very interesting idea. The discussion about the
`Individual`
vs
`Global` checkpoint was raised before, but the main concern
was
from
two
aspects:
- Non-deterministic replaying may lead to an inconsistent
view
of
checkpoint
- It is not easy to form a clear cut of past and future and
hence no
clear
cut of where the start point of the source should begin to
replay
from.
Starting from independent subgraphs as you proposed may be a
good
starting
point. However, when we talk about subgraph, do we mention it
as
a
job
subgraph (each vertex is one or more operators) or execution
subgraph
(each
vertex is a task instance)?
If it is a job subgraph, then indeed, why not separate it
into
multiple
jobs as Caizhi mentioned.
If it is an execution subgraph, then it is difficult to
handle
rescaling
due to inconsistent views of checkpoints between tasks of the
same
operator.
`Individual/Subgraph Checkpointing` is definitely an
interesting
direction
to think of, and I'd love to hear more from you!
Best,
Yuan
On Mon, Feb 7, 2022 at 10:16 AM Caizhi Weng <
tsreape...@gmail.com>
wrote:
Hi Gyula!
Thanks for raising this discussion. I agree that this will
be
an
interesting feature but I actually have some doubts about
the
motivation
and use case. If there are multiple individual subgraphs in
the
same
job,
why not just distribute them to multiple jobs so that each
job
contains
only one individual graph and can now fail without
disturbing
the
others?
Gyula Fóra <gyf...@apache.org> 于2022年2月7日周一 05:22写道：
Hi all!
At the moment checkpointing only works for healthy jobs
with
all
running
(or some finished) tasks. This sounds reasonable in most
cases
but
there
are a few applications where it would make sense to
checkpoint
failing
jobs
as well.
Due to how the checkpointing mechanism works, subgraphs
that
have a
failing
task cannot be checkpointed without violating the
exactly-once
semantics.
However if the job has multiple independent subgraphs
(that
are
not
connected to each other), even if one subgraph is
failing,
the
other
completely running one could be checkpointed. In these
cases
the
tasks
of
the failing subgraph could simply inherit the last
successful
checkpoint
metadata (before they started failing). This logic would
produce
a
consistent checkpoint.
The job as a whole could now make stateful progress even
if
some
subgraphs
are constantly failing. This can be very valuable if for
some
reason
the
job has a larger number of independent subgraphs that are
expected
to
fail
every once in a while, or if some subgraphs can have
longer
downtimes
that
would now cause the whole job to stall.
What do you think?
Cheers,
Gyula
