flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Vijay Balakrishnan <bvija...@gmail.com>
Subject Timestamp Watermark Assigner bpund question
Date Tue, 09 Apr 2019 23:41:22 GMT
Hi,
I have created a TimestampAssigner as follows.
I want to use monitoring.getEventTimestamp() with an Event Time processing
and collected aggregated stats over time window intervals of 5 secs, 5 mins
etc. Is this the right way to create the TimeWaterMarkAssigner with a bound
? I want to collect the stats for each eventTimestamp + window intervals.
My understanding - *the generated watermark which is eventTimestamp + bound
will collect all the eventTimestamp's which arrive within that Watermark
inside each eventTimestamp + 5s etc window interval. Or does this bound
have to be based on the windowInterval i.e extractedTimestamp +
windowInterval + bound *??


> *public class MonitoringTSWAssigner implements
> AssignerWithPunctuatedWatermarks<Monitoring> {*
> * private long bound = 5 * (long) 1000; *
> * public long extractTimestamp(Monitoring monitoring, long previousTS) {*
> *        return monitoring.getEventTimestamp();**    }*
>
> *    public Watermark checkAndGetNextWatermark(Monitoring monitoring, long
> extractedTimestamp) {*
> *        return new Watermark(extractedTimestamp + bound);//<==== should
> it be - bound ?*
> *    }**}*


Used here:

> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
> final DataStreamSource<Monitoring> monitoringDataStreamSource =
> env.addSource(....);
> DataStream<Monitoring> kinesisStream =
> monitoringDataStreamSource.assignTimestampsAndWatermarks(new
> MonitoringTSWAssigner());
> KeyedStream<Monitoring, Tuple> monitoringTupleKeyedStream =
> kinesisStream.keyBy("deployment", .....);
> final WindowedStream<Monitoring, Tuple, TimeWindow> windowStream =
>                 monitoringTupleKeyedStream.timeWindow(Time.seconds(5));//5
> sec time window


TIA,

Mime
View raw message