flink教程-3.时间戳和watermark管理

Flink时间戳和watermarkfinalStreamExecutionEnvironmentenvfinalStreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnviro分配时间事件只有带着时间戳，同时flink能到该时间戳，才能使用事件时间进行处时间戳分配和watermark生成密切相关，watermark可以告诉系统基于事件时间分配时间戳和产生watermark通过使用时间戳分配器/watermark：flink时间戳分配器同时也会负责生成watermark。时间戳和watermark1970-01-01T00:00:00Z带时间戳和watermark的源数据流的源函数能够直接分配时间戳给事件，同watermark。当使用这sourcewatermark。sourcesourcesourceContextcollectWithTimestamp(...watermark，必须使用SourceContextemitWatermark(Watermark)方法。while(/*condition*/){MyTypewhile(/*condition*/){MyTypenext=ctx.collectWithTimestamp(next,ifif(next.hasWatermarkTime())}}}时间戳分配器watermarkwatermark，mapfliterwindowsKafkaflinksource或者consumerfinalStreamExecutionEnvironmentfinalStreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnviroDataStream<MyEvent>stream=myFormat,myFilePath,FileProcessingMode.PROCESS_CONTINUOUSLY,100,FilePathFilter.createDefaultFilter(),typeInfo);DataStream<MyEvent>withTimestampsAndWatermarks=.filter(event->event.severity()==WARNING.keyBy((event)->event.getGroup().reduce((a,b)->a.add(b)使用周期性的watermarkwatermark。下面两个时间戳分配器都是会周期性的生成watermark。注意maxOutOfOrderness毫秒，时间都基于事件时间。publicclassBoundedOutOfOrdernessGeneratorimplementspublicclassBoundedOutOfOrdernessGeneratorimplementsAssignerWithPeriodicWatermarks<MyEvent>{privatefinallongmaxOutOfOrderness=3500;//3.5 publiclongextractTimestamp(MyEven ement,longpreviousElementTimestamp){longtimestamp=element.getCreationTime();currentMaxTimestamp=Math.max(timestamp,currentMaxTimestamp);returntimestamp;}publicWatermarkgetCurrentWatermark()//returnthewatermarkascurrenthighesttimestampminustheout-of-sreturnreturnnewWatermark(currentMaxTimestamp-}}Watermark(t)=MyEvent>privatefinallongmaxTimeLag=5000;//5publiclongextractTimestamp(MyEventelement,longp)return}publicWatermarkgetCurrentWatermark()//returnthewatermarkascurrenttimeminusumtimeg}}Puncturated可以通AssignerWithPunctuatedWatermarkswatermark，其有个extractTimestampcheckAndGetNextWatermark有两个参数，一个是调用的事件，一个是extractTimestamp时间戳，可以决定是否产watermark。watermark发射。publicclassPunctuatedAssignerpublicclassPunctuatedAssignerimplements{@Overridepubliclong ement,longextractedTimestamp){return watermark产生额外的计算，所以过多watermarkKafka分区的时间当使用kafka作为数据源的话，每个kafka分区都可以满足一个很简单的时间模式(时间戳递增或者有界的无序)。但是，当消费kafka数据的时候，往往会例如，如果单kafka时间戳严格的递增，那么使用递增时间watermark产生的watermark就会是个很完美的全局watermark下图展示了如何使用kafka单分区，和这种情况下watermark是如何在DataStreamDataStream<MyType>stream=publiclongextra cendingTimestamp(MyTypeelement){returnelement.eventTimestamp();}",