flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Narayanan Arunachalam <narayanan.arunacha...@gmail.com>
Subject Re: TimerService/Watermarks and Checkpoints
Date Wed, 30 May 2018 21:56:53 GMT
Thanks for the explanation. I looked at this metric closely and noticed
there are some events arriving in out of order. The hypothesis I have is,
when the job is restarted, all of the small out of order chunks add up and
show a significant number. The graph below shows the number of out of order
events every min. The job was started with new state at 11:53 am and then
restarted with the previous checkpoint at 1:24 pm.

That said, after restart the out of order events number is very high though
:thinking_face:





On Wed, May 30, 2018 at 1:55 PM, Fabian Hueske <fhueske@gmail.com> wrote:

> Hi Nara and Sihua,
>
> That's indeed an unexpected behavior and it would be good to identify the
> reason for the late data.
>
> As Sihua said, watermarks are currently not checkpointed and reset to
> Long.MIN_VALUE upon restart.
> AFAIK, the main reason why WMs are not checkpointed is that the special
> type of operator state that is required for this (union-list state) wasn't
> available when the mechanism was implemented.
> I think there are plans to address this shortcoming (see FLINK-5601 [1]).
>
> Best, Fabian
>
> [1] https://issues.apache.org/jira/browse/FLINK-5601
>
> 2018-05-30 19:00 GMT+02:00 Narayanan Arunachalam <
> narayanan.arunachalam@gmail.com>:
>
>> Thanks Sihua. If it's reset to Long.MIN_VALUE I can't explain why
>> outOfOrderEvents are reported. Because the event time on the data will
>> always be greater than Long.MIN_VALUE.
>>
>> Following are the steps to reproduce this scenario.
>> - A source to produce events with timestamps that is increasing for every
>> event produced
>> - Use TimeCharacteristic.EventTime
>> - Use BoundedOutOfOrdernessTimestampExtractor with maxOutOfOrderness set
>> to 60s.
>> - Enable checkpoints
>> - ProcessFunction impl to report a counter to some metrics backend when
>> the timestamp of the event is less than currentWatermark
>> - No out of order events will be reported initially. After few
>> checkpoints are created, cancel and restart the job from a previous
>> checkpoint.
>>
>> *Note*: The event stream really doesn't have out of order data. Job
>> restart from a checkpoint causes this artificial out of order events
>> because of the watermark value.
>>
>> Regards,
>> Nara
>>
>>
>>
>>
>> On Tue, May 29, 2018 at 7:54 PM, sihua zhou <summerleafs@163.com> wrote:
>>
>>> Hi Nara,
>>>
>>> yes, the watermark in TimerService is not covered by the checkpoint,
>>> everytime the job is restarted from a previous checkpoint, it is reset to
>>> Long.MIN_VALUE. I can see it a bit tricky to cover it into the checkpoint,
>>> especially when we need to support rescaling(it seems not like a purely
>>> keyed or a operate state), maybe @Stefan or @Aljoscha could give you more
>>> useful information about why it wasn't covered by the checkpoint.
>>>
>>> Best, Sihua
>>>
>>>
>>>
>>> On 05/30/2018 05:44,Narayanan Arunachalam<narayanan.arunacha
>>> lam@gmail.com> <narayanan.arunachalam@gmail.com> wrote:
>>>
>>> Hi,
>>>
>>> Is it possible the watermark in TimerService not getting reset when a
>>> job is restarted from a previous checkpoint? I would expect the watermark
>>> in a TimerService also to go back in time.
>>>
>>> I have the following ProcessFunction implementation.
>>>
>>>   override def processElement(
>>>     e: TraceEvent,
>>>     ctx: ProcessFunction[
>>>       TraceEvent,
>>>       Trace
>>>     ]#Context,
>>>     out: Collector[Trace]
>>>   ): Unit = {
>>>
>>>     if (e.getAnnotationTime() < ctx.timerService().currentWatermark()) {
>>>       registry.counter("tracing.outOfOrderEvents").increment()
>>>     } else {
>>>     ....
>>>     }
>>>
>>> I am noticing the outOfOrderEvents are reported until new events are
>>> read in to the stream since the last restart.
>>>
>>> Regards,
>>> Nara
>>>
>>>
>>
>

Mime
View raw message