Hi Cliff!
Sorry to hear that you are running into so much trouble with the
checkpointing.
Here are a few things we found helping another user that was running into a
similar issue:
(1) Make sure you use a rather new version of the 1.2 branch (there have
been some important fixes added to the network in the past weeks)
(2) Use TaskManagers with one slot (start more TaskManagers instead,
multiple TaskManagers can run on the same machine). That way, network
connections do not get into each other's way. Flink's connection
multiplexing has a vulnerability that may slow down networks in some cases,
which can affect checkpoints.
(3) This is very important: Try to move all state that is large to keyed
state (value state, list state, reducing state, ...) and use RocksDB as the
state backend. That way snapshots happen asynchronously, which greatly
helps the checkpoints.
(4) Briefly looking over the File Reading Operator code, it could be that
the operator does not release the checkpoint lock long enough for barriers
to get injected in a timely fashion. If you are running a custom Flink
build anyways, you could try to change the following code block in the
"ContinuousFileReaderOperator" (as an experiment, not a final fix)
while (!format.reachedEnd()) {
synchronized (checkpointLock) {
nextElement = format.nextRecord(nextElement);
if (nextElement != null) {
readerContext.collect(nextElement);
} else {
break;
}
}
}
while (!format.reachedEnd()) {
synchronized (checkpointLock) {
nextElement = format.nextRecord(nextElement);
if (nextElement != null) {
readerContext.collect(nextElement);
} else {
break;
}
}
// give the checkpoint thread a change to work
Thread.yield();
}
If you are already doing all of that, or if this does not help, we have two
courses of action (if you can bear with us for a bit):
- There is some much enhanced Checkpoint Monitoring almost ready to be
merged. That should help in fining out where the barriers get delayed.
- Finally, we are experimenting with some other checkpoint alignment
variants (alternatives to the BarrierBuffer). We can ping you when we have
an experimental branch with that.
Hope we get this under control!
Best,
Stephan
On Fri, Dec 23, 2016 at 3:47 PM, Cliff Resnick <cre...@gmail.com> wrote:
> We are running a DataStream pipeline using Exactly Once/Event Time
> semantics on 1.2-SNAPSHOT. The pipeline sources from S3 using the
> ContinuousFileReaderOperator. We use a custom version of the
> ContinuousFileMonitoringFunction since our source directory changes over
> time. The pipeline transforms and aggregates tuples of data that is steady
> over time (spike-less), windowing by hour with an allowed lateness of 6
> hours. We are running on ec2 c4 instances in a simple YARN setup.
>
> What we are seeing is that as we scale the system to hundreds of cores
> reading around 10 million events per second, the pipeline might checkpoint
> up to few times before it reaches a state where it becomes unable to
> complete a checkpoint. The checkpoint interval does not seem to matter as
> we've tested intervals from 5 minutes to one hour and timeouts up to one
> hour as well.
> What we'll usually see is something like:
>
> checkpoint 1 (2G) 1 minute
> checkpoint 2 (3-4G) 1-5 minutes
> checkpoint 3 (5-6G) 1-6 minutes
> checkpoint 4-5 (mixed bag, 4-25 minutes, or never)
> checkpoint 6-n never
>
> We've added debugging output to Flink internal code, e.g.
> BarrierBuffer.java. From the debugging output it's clear that the actual
> checkpoint per operator always completes in at most several seconds. What
> seem to be happening, however, is that CheckpointBarriers start to become
> slower to arrive, and after a few checkpoints it gets worse with
> CheckpointBarriers going greatly askew and finally never arriving.
> Meanwhile we can see that downstream counts are closely tailing upstream
> counts, < .01 behind, but once the barrier flow seemingly stops, the
> downstream slows to a stop as (I guess) network buffers fill, then the
> pipeline is dead.
>
> Meanwhile, cluster resources are not being stressed.
>
> At this point we've stripped down the pipeline, tried various StateBackend
> configs, etc. but the result is invariably the same sad story. It would be
> great if somebody could provide more insight into where things might be
> going wrong. Hopefully this is a simple config issue, but we'd be open to
> any and all suggestions regarding testing, tweaking, etc.
>
> -Cliff
>
>
>
>
>
>
