flink watermark介绍-白红宇

flink watermark介绍

阅读量：5907 次

发布时间：2019-06-19

本文共 9708 字，大约阅读时间需要 32 分钟。

转发请注明原创地址 http://www.cnblogs.com/dongxiao-yang/p/7610412.html

一概念

watermark是flink为了处理eventTime窗口计算提出的一种机制,本质上也是一种时间戳，由flink souce或者自定义的watermark生成器按照需求定期或者按条件生成一种系统event，与普通数据流event一样流转到对应的下游operations，接收到watermark数据的operator以此不断调整自己管理的window event time clock。

( A watermark is a special event signaling that time in the event stream (i.e., the real-world timestamps in the event stream) has reached a certain point (say, 10am), and thus no event with timestamp earlier than 10am will arrive from now on. These watermarks are part of the data stream alongside regular events, and a Flink operator advances its event time clock to 10am once it has received a 10am watermark from all its upstream operations/sources)

二 TimestampAssigner和Watermark

首先，eventTime计算意味着flink必须有一个地方用于抽取每条消息中自带的时间戳，所以TimestampAssigner的实现类都要具体实现

long extractTimestamp(T element, long previousElementTimestamp);方法用来抽取当前元素的eventTime，这个eventTime会用来决定元素落到下游的哪个或者哪几个window中进行计算。

其次，在数据进入window前，需要有一个Watermarker生成当前的event time对应的水位线，flink支持两种后置的Watermarker：Periodic和Punctuated，一种是定期产生watermark（即使没有消息产生），一种是在满足特定情况的前提下触发。两种Watermark分别需要实现接口为

Watermark getCurrentWatermark()和Watermark checkAndGetNextWatermark(T lastElement, long extractedTimestamp);

帖几个官网给出的实现样例

Periodic Watermarks

1 /** 2  * This generator generates watermarks assuming that elements arrive out of order, 3  * but only to a certain degree. The latest elements for a certain timestamp t will arrive 4  * at most n milliseconds after the earliest elements for timestamp t. 5  */ 6 public class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks
     
       { 7  8     private final long maxOutOfOrderness = 3500; // 3.5 seconds 9 10     private long currentMaxTimestamp;11 12     @Override13     public long extractTimestamp(MyEvent element, long previousElementTimestamp) {14         long timestamp = element.getCreationTime();15         currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);16         return timestamp;17     }18 19     @Override20     public Watermark getCurrentWatermark() {21         // return the watermark as current highest timestamp minus the out-of-orderness bound22         return new Watermark(currentMaxTimestamp - maxOutOfOrderness);23     }24 }25 26 /**27  * This generator generates watermarks that are lagging behind processing time by a fixed amount.28  * It assumes that elements arrive in Flink after a bounded delay.29  */30 public class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks
      
        {31 32     private final long maxTimeLag = 5000; // 5 seconds33 34     @Override35     public long extractTimestamp(MyEvent element, long previousElementTimestamp) {36         return element.getCreationTime();37     }38 39     @Override40     public Watermark getCurrentWatermark() {41         // return the watermark as current time minus the maximum time lag42         return new Watermark(System.currentTimeMillis() - maxTimeLag);43     }44 }

Punctuated Watermarks

1 public class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks
     
       { 2  3     @Override 4     public long extractTimestamp(MyEvent element, long previousElementTimestamp) { 5         return element.getCreationTime(); 6     } 7  8     @Override 9     public Watermark checkAndGetNextWatermark(MyEvent lastElement, long extractedTimestamp) {10         return lastElement.hasWatermarkMarker() ? new Watermark(extractedTimestamp) : null;11     }12 }

三代码调试

1 public class WindowWaterMark {  2   3     public static void main(String[] args) throws Exception {  4         // TODO Auto-generated method stub  5   6   7         String hostName = "localhost";  8         Integer port = Integer.parseInt("8001");  9  10         // set up the execution environment 11         final StreamExecutionEnvironment env = StreamExecutionEnvironment 12                 .getExecutionEnvironment(); 13         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 14         env.setParallelism(2); 15          env.getConfig().setAutoWatermarkInterval(9000); 16  17         // get input data 18         DataStream
     
       text = env.socketTextStream(hostName, port); 19  20         DataStream
      
       
        > counts = 21      22         text.filter(new FilterClass()).map(new LineSplitter()) 23         .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks
        
         
          >(){ 24  25             private long  currentMaxTimestamp = 0l; 26             private final long maxOutOfOrderness = 10000l; 27              28             @Override 29             public long extractTimestamp(Tuple3
          
            element, 30 long previousElementTimestamp) { 31 // TODO Auto-generated method stub 32 long timestamp= element.f1; 33 currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp); 34 System.out.println("get timestamp is "+timestamp+" currentMaxTimestamp "+currentMaxTimestamp); 35 return timestamp; 36 } 37 38 @Override 39 public Watermark getCurrentWatermark() { 40 // TODO Auto-generated method stub 41 System.out.println("wall clock is "+ System.currentTimeMillis() +" new watermark "+(currentMaxTimestamp - maxOutOfOrderness)); 42 return new Watermark(currentMaxTimestamp - maxOutOfOrderness); 43 44 } 45 46 }) 47 .keyBy(0) 48 .timeWindow(Time.seconds(20)) 49 // .allowedLateness(Time.seconds(10)) 50 .sum(2); 51 52 counts.print(); 53 54 // execute program 55 env.execute("Java WordCount from SocketTextStream Example"); 56 } 57 58 59 public static final class LineSplitter implements 60 MapFunction
           
            > { 61 62 @Override 63 public Tuple3
            
              map(String value) throws Exception { 64 // TODO Auto-generated method stub 65 String[] tokens = value.toLowerCase().split("\\W+"); 66 67 long eventtime = Long.parseLong(tokens[1]); 68 69 return new Tuple3
             
              (tokens[0], eventtime, 1); 70 } 71 } 72 73 private static class MyTimestamp extends 74 AscendingTimestampExtractor
              
               
                > { 75 private static final long serialVersionUID = 1L; 76 77 @Override 78 public long extractAscendingTimestamp( 79 Tuple3
                
                  element) { 80 // TODO Auto-generated method stub 81 return element.f1; 82 } 83 84 } 85 86 87 public static final class FilterClass implements FilterFunction
                 
                   88 { 89 90 @Override 91 public boolean filter(String value) throws Exception { 92 // TODO Auto-generated method stub 93 if(StringUtils.isNullOrWhitespaceOnly(value)) 94 { 95 return false; 96 } 97 else 98 { 99 return true;100 }101 }102 103 }104 105 }

测试代码如上，注意这段代码手动更改了autowatermarkinterval的时间为9s以便于观察方法调用顺序。

首先启动job不输入数据，30s后日志输出为

wall clock is 1506680562679 new watermark -10000

wall clock is 1506680571683 new watermark -10000

wall clock is 1506680580687 new watermark -10000

.........................

这说明在没有数据输入的情况下PeriodicWatermarks仍然会周期性调用getCurrentWatermark这个方法，每次有两条相同wall clock的日志跟程序里env.setParallelism(2)这个参数相同，表明watermark与operator的并发一致。

输入数据aaaa 1506590035000

日志输出为

wall clock is 1506681868124 new watermark -10000

wall clock is 1506681877129 new watermark -10000

get timestamp is 1506590035000 currentMaxTimestamp 1506590035000

wall clock is 1506681886132 new watermark 1506590025000

wall clock is 1506681886132 new watermark -10000

wall clock is 1506681895136 new watermark -10000

wall clock is 1506681895136 new watermark 1506590025000

...........................................

上述日志表明接收到消息后extractTimestamp这个方法会被立即调用,但是同时注意到wall clock日志的打印时间完全没有受到数据流入的影响，所以在PeriodicWatermarks这个是线下，watermark的产生时间和速率与数据流的输入无关。

需要说明的是，时间窗口的起始时间计算方法为

public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {
     return timestamp - (timestamp - offset + windowSize) % windowSize; }

所有对于上述测试代码里时间长度为20s的滚动窗口，默认下，在每分钟内窗口的起止时间都是（0~20）（20~40）（40~60）这样，我们的第一条数据aaaa 1506590035000对应时间为2017/9/28 17:13:55，所以它将会在（2017/9/28 17:13:40~2017/9/28 17:14:00）这个窗口完成计算

继续输入数据

cc 1506590035000

bb 1506590035000

aaaa 1506590035000

bb 1506590035000

aaaa 1506590041000 //上调数据的eventTime至2017/9/28 17:14:01，超过前一个window的结束时间

bb 1506590041000

cc 1506590041000

日志输出为

get timestamp is 1506590041000 currentMaxTimestamp 1506590041000

wall clock is 1507522499419 new watermark 1506590031000

wall clock is 1507522499424 new watermark 1506590025000