@vinay - Is it in your case large state that causes slower checkpoints? On Mon, Sep 26, 2016 at 6:17 PM, vinay patil <vinay18.pa...@gmail.com> wrote:
> Hi, > > I am also facing this issue, in my case the data is flowing continuously > from the Kafka source, when I increase the checkpoint interval to 60000, > the data gets written to S3 sink. > > Is it because some operator is taking more time for processing, like in my > case I am using a time window of 1sec. > > Regards, > Vinay Patil > > On Mon, Sep 26, 2016 at 10:08 AM, Chakravarthy varaga [via Apache Flink > User Mailing List archive.] <[hidden email] > <http:///user/SendEmail.jtp?type=node&node=9179&i=0>> wrote: > >> Hi Stefan, >> >> Please find my responses below. >> >> - What source are you using for the slow input? >> * [CVP] - Both stream as pointed out in my first mail, are Kafka >> Streams* >> - How large is the state that you are checkpointing? >> >> *[CVP] - I have enabled checkpointing on the StreamEnvironment as below.* >> >> >> >> * final StreamExecutionEnvironment streamEnv = >> StreamExecutionEnvironment.getExecutionEnvironment(); >> streamEnv.setStateBackend(new >> FsStateBackend("file:///tmp/flink/checkpoints")); >> streamEnv.enableCheckpointing(10000);* >> >> >> * In terms of the state stored, the KS1 stream has payload of 100K >> events/second, while KS2 have about 1 event / 10 minutes... basically the >> operators perform flatmaps on 8 fields of tuple (all fields are >> primitives). If you look at the states' sizes in dashboard they are in >> Kb...* >> - 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? >> [CVP] -There are no back pressure atleast from the sample >> computation in the flink dashboard. 100K/second is low load for flink's >> benchmarks. I could not quite get the barriers vs snapshot state. I have >> attached the Task Manager log (DEBUG) info if that will interest you. >> >> I have attached the checkpoints times' as .png from the dashboard. >> Basically if you look at checkpoint IDs 28 & 29 &30- you'd see that the >> checkpoints take more than a minute in each case. Before these checkpoints, >> the KS2 stream did not have any events. As soon as an event(should be in >> bytes) was generated, the checkpoints went slow and subsequently a minute >> more for every checkpoint thereafter. >> >> This log was collected from the standalone flink cluster with 1 job >> manager & 2 TMs. 1 TM was running this application with checkpointing >> (parallelism=1) >> >> Please let me know if you need further info., >> >> >> >> On Fri, Sep 23, 2016 at 6:26 PM, Stephan Ewen <[hidden email] >> <http:///user/SendEmail.jtp?type=node&node=9176&i=0>> wrote: >> >>> 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 <[hidden email] >>> <http:///user/SendEmail.jtp?type=node&node=9176&i=1>> 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 <[hidden email] >>>> <http:///user/SendEmail.jtp?type=node&node=9176&i=2>>: >>>> >>>>> 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 >>>>> >>>> >>>> >>> >> >> *flink_job_Plan.png* (42K) Download Attachment >> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/attachment/9176/0/flink_job_Plan.png> >> *Flink-Checkpoint-Times.png* (65K) Download Attachment >> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/attachment/9176/1/Flink-Checkpoint-Times.png> >> *flink-qchavar-taskmanager-1-elxa1h67k32.log* (442K) Download Attachment >> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/attachment/9176/2/flink-qchavar-taskmanager-1-elxa1h67k32.log> >> >> >> ------------------------------ >> If you reply to this email, your message will be added to the discussion >> below: >> http://apache-flink-user-mailing-list-archive.2336050.n4. >> nabble.com/Flink-Checkpoint-runs-slow-for-low-load-stream- >> tp9147p9176.html >> To start a new topic under Apache Flink User Mailing List archive., email >> [hidden >> email] <http:///user/SendEmail.jtp?type=node&node=9179&i=1> >> To unsubscribe from Apache Flink User Mailing List archive., click here. >> NAML >> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml> >> > > > ------------------------------ > View this message in context: Re: Flink Checkpoint runs slow for low load > stream > <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Checkpoint-runs-slow-for-low-load-stream-tp9147p9179.html> > Sent from the Apache Flink User Mailing List archive. mailing list archive > <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/> at > Nabble.com. >
