Hi!
Let's try to figure that one out. Can you give us a bit more information?
- What source are you using for the slow input?
- How large is the state that you are checkpointing?
- Can you try to see in the log if actually the state snapshot takes that
long, or if it simply takes long for the checkpoint barriers to travel
through the stream due to a lot of backpressure?
Greetings,
Stephan
On Fri, Sep 23, 2016 at 3:35 PM, Fabian Hueske <fhue...@gmail.com> wrote:
> Hi CVP,
>
> I'm not so much familiar with the internals of the checkpointing system,
> but maybe Stephan (in CC) has an idea what's going on here.
>
> Best, Fabian
>
> 2016-09-23 11:33 GMT+02:00 Chakravarthy varaga <chakravarth...@gmail.com>:
>
>> Hi Aljoscha & Fabian,
>>
>> I have a stream application that has 2 stream source as below.
>>
>> KeyedStream<String, String> *ks1* = ds1.keyBy("*") ;
>> KeyedStream<Tuple2<String, V>, String> *ks2* = ds2.flatMap(split T
>> into k-v pairs).keyBy(0);
>>
>> ks1.connect(ks2).flatMap(X);
>> //X is a CoFlatMapFunction that inserts and removes elements from
>> ks2 into a key-value state member. Elements from ks1 are matched against
>> that state. the CoFlatMapFunction operator maintains
>> ValueState<Tuple2<Long, Long>>;
>>
>> //ks1 is streaming about 100K events/sec from kafka topic
>> //ks2 is streaming about 1 event every 10 minutes... Precisely when
>> the 1st event is consumed from this stream, checkpoint takes 2 minutes
>> straight away.
>>
>> The version of flink is 1.1.2.
>>
>> I tried to use checkpoint every 10 Secs using a FsStateBackend... What I
>> notice is that the checkpoint duration is almost 2 minutes for many cases,
>> while for the other cases it varies from 100 ms to 1.5 minutes frequently.
>> I'm attaching the snapshot of the dashboard for your reference.
>>
>> Is this an issue with flink checkpointing?
>>
>> Best Regards
>> CVP
>>
>
>
