flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Saiph Kappa <saiph.ka...@gmail.com>
Subject Re: Why tuples are not ignored after watermark?
Date Thu, 15 Sep 2016 18:17:42 GMT
And is it possible to share state across parallel instances with
AssignerWithPunctuatedWatermarks?

Thanks!

On Wed, Sep 14, 2016 at 9:52 AM, Aljoscha Krettek <aljoscha@apache.org>
wrote:

> Hi,
> the problem might be that your timestamp/watermark assigner is run in
> parallel and that only one parallel instance of those operators emits the
> watermark because only one of those parallel instances sees the element
> with _3 == 9000. For the watermark to advance at an operator it needs to
> advance in all upstream operations.
>
> Cheers,
> Aljoscha
>
> On Fri, 9 Sep 2016 at 18:29 Saiph Kappa <saiph.kappa@gmail.com> wrote:
>
>> Hi,
>>
>> I have a streaming (event time) application where I am receiving events
>> with the same assigned timestamp. I receive 10000 events in total on a
>> window of 5 minutes, but I emit water mark when 9000 elements have been
>> received. This watermark is 6 minutes after the assigned timestamps. My
>> question is: why the function that is associated with the window reads
>> 10000 elements and not 9000? All elements that have a timestamp lower than
>> the watermark should be ignored (1000), but it's not happening.
>>
>> Here is part of the code:
>> «
>> val env = StreamExecutionEnvironment.getExecutionEnvironment
>> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
>> val rawStream = env.socketTextStream("localhost", 4321)
>>
>> val punctuatedAssigner = new AssignerWithPunctuatedWatermarks[(String,
>> Int, Long)] {
>>       val timestamp = System.currentTimeMillis();
>>
>>       override def extractTimestamp(element: (String, Int, Long),
>> previousElementTimestamp: Long): Long =
>>         timestamp
>>
>>       override def checkAndGetNextWatermark(lastElement: (String, Int,
>> Long), extractedTimestamp: Long): Watermark = {
>>         if(lastElement._3 == 9000) {
>>           val ts = extractedTimestamp + TimeUnit.MINUTES.toMillis(6)
>>           new watermark.Watermark(ts)
>>         } else null
>>       }
>>     }
>>
>> val stream = rawStream.map(line => {
>>       val Array(p1, p2, p3) = line.split(" ")
>>       (p1, p2.toInt, p3.toLong)
>>     })
>>       .assignTimestampsAndWatermarks(punctuatedAssigner)
>>
>> stream.keyBy(1).timeWindow(Time.of(5, TimeUnit.MINUTES)).apply(function)
>> »
>>
>> Thanks!
>>
>

Mime
View raw message