关于event-time的定义与产生时间戳位置的问题。

Hi, 可以看下事件时间戳的生成，https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/event_timestamps_watermarks.html 下面例子里时间戳都是来自element里面的时间字段。还有一个AscendingTimestampExtractor。 /** * This generator generates watermarks assuming that elements arrive out of order, * but only to a certain degree. The latest elements for a certain timestamp t will arrive * at most n milliseconds after the earliest elements for timestamp t. /publicclassBoundedOutOfOrdernessGeneratorimplementsAssignerWithPeriodicWatermarks {privatefinallongmaxOutOfOrderness=3500;// 3.5 secondsprivatelongcurrentMaxTimestamp;@OverridepubliclongextractTimestamp(MyEventelement,longpreviousElementTimestamp){longtimestamp=element.getCreationTime();currentMaxTimestamp=Math.max(timestamp,currentMaxTimestamp);returntimestamp;}@OverridepublicWatermarkgetCurrentWatermark(){// return the watermark as current highest timestamp minus the out-of-orderness boundreturnnewWatermark(currentMaxTimestamp-maxOutOfOrderness);}}/ * * This generator generates watermarks that are lagging behind processing time by a fixed amount. * It assumes that elements arrive in Flink after a bounded delay. */publicclassTimeLagWatermarkGeneratorimplementsAssignerWithPeriodicWatermarks {privatefinallongmaxTimeLag=5000;// 5 seconds@OverridepubliclongextractTimestamp(MyEventelement,longpreviousElementTimestamp){returnelement.getCreationTime();}@OverridepublicWatermarkgetCurrentWatermark(){// return the watermark as current time minus the maximum time lagreturnnewWatermark(System.currentTimeMillis()-maxTimeLag);}} publicclassPunctuatedAssignerimplementsAssignerWithPunctuatedWatermarks {@OverridepubliclongextractTimestamp(MyEventelement,longpreviousElementTimestamp){returnelement.getCreationTime();}@OverridepublicWatermarkcheckAndGetNextWatermark(MyEventlastElement,longextractedTimestamp){returnlastElement.hasWatermarkMarker()?newWatermark(extractedTimestamp):null;}} 希望能有所帮助。

DataStream和kafkaSource后面都可以调用assignTimestampsAndWatermarks。

kafkaSource.assignTimestampsAndWatermarks(newAscendingTimestampExtractor (){@OverridepubliclongextractAscendingTimestamp(MyTypeelement){returnelement.eventTimestamp();}}); *来自志愿者整理的flink邮件归档