flink-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Robert Metzger <rmetz...@apache.org>
Subject Re: Playing with EventTime in DataStreams
Date Thu, 25 Feb 2016 19:05:41 GMT
Hi,

I had a similar issue recently.
Instead of
 input.assignTimestampsAndWatermarks

you have to do:

 input = input.assignTimestampsAndWatermarks

On Thu, Feb 25, 2016 at 6:14 PM, Nam-Luc Tran <namluc.tran@euranova.eu>
wrote:

> Hello everyone,
>
> I am currently playing with streams which timestamp is defined by
> EventTime. I currently have the following code:
>
>       final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
>
> env.getConfig().enableTimestamps();//.setAutoWatermarkInterval(10000);
>       env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>
>       DataStream<String> input =
> env.readTextFile("file:///var/log/syslog");
>       input.assignTimestampsAndWatermarks(new
> AssignTimestampFromLogEvent());
>
>       input.timeWindowAll(Time.minutes(5)).apply(new
> AllWindowFunction<Iterable<String>, String, TimeWindow>() {
>          @Override
>          public void apply(TimeWindow window, Iterable<String> values,
> Collector<String> out) throws Exception {
>             for(String t:values) {
>                out.collect(t);
>             }
>          }
>       }).print();
>
> (...)
>
> public static final class AssignTimestampFromLogEvent extends
> AscendingTimestampExtractor<String> {
>    @Override
>    public long extractAscendingTimestamp(String element, long
> previousElementTimestamp){
>       String date = element.substring(0,15);
>       SimpleDateFormat sdf = new SimpleDateFormat("MMM dd HH:mm:ss");
>       Date ddate = null;
>       try {
>          ddate = sdf.parse(date);
>       } catch (ParseException e) {
>          e.printStackTrace();
>       }
>       return ddate.getTime();
>    }
> }
>
>
> What I expect it to do is to read the syslog, assign timestamp and do
> 5 minutes windows *based on the syslog event time*, as I've configured
> the stream to do it. It however does not do that, and does the windows
> based on processing time.
>
> What am I missing here?
>
> Best regards,
>
> --
>
> *Nam-Luc TRAN*
>
> R&D Manager
>
> EURA NOVA
>
> (M) +32 498 37 36 23
>
> *euranova.eu <http://euranova.eu>*
>

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message