Flink中的时间语义与Watermark概念

浅浅的花香味﹌ 2023-01-15 14:21 218阅读 0赞

## 一、时间语义 ##

### 1.1 时间语义类型 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70]

*  Event Time：事件创建的时间
 *  Ingestion Time：数据进入Flink的时间
 *  Processing Time：执行操作算子的本地系统时间，与机器相关

**问题：哪种时间语义更重要？**

不同的时间语义有不同的应用场合，通常更关心的是**事件时间**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 1]  
某些应用场合，不应该使用Processing Time。Event Time可以从日志数据的时间戳（timestamp）中提取  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 2]

### 1.2 实际应用 ###

## 二、水位线（Watermark） ##

**乱序数据的影响**

*  当 Flink 以 Event Time 模式处理数据流时，它会根据数据里的时间戳来处理基于时间的算子
 *  由于网络、分布式等原因，会导致乱序数据的产生  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 3]

**怎样避免乱序数据带来的计算不正确问题呢？**

遇到一个时间戳达到了窗口关闭时间，不应该立刻触发窗口计算，而是等待一段时间，等迟到的数据来了再关闭窗口

*  Watermark是一种衡量Event Time进展的机制，可以设定**延迟触发**
 *  Watermark是用于处理乱序事件的，处理乱序事件正确的方法，通常是用Watermark机制结合window来实现
 *  数据流中的Watermark用于表示 **timestamp小于Watermark** 的数据，都已经到达了。因此，window的执行也是由Watermark触发的
 *  watermark用来让程序自己平衡延迟和结果正确性

### 2.1 Flink三种方法保证数据准确性（三重保证） ###

（1）Watermark，可以保证 **几百毫秒内** 的乱序数据的准确性  
（2）在（1）的基础上，可以再使用 **allowedLateness** 设置等待时间  
（3）在（2）的基础上，可以再使用**侧输出流**

### 2.2 Watermark的特点 ###

*  Watermark是一条特殊的数据记录
 *  Watermark必须**单调递增**，以确保任务的事件时间时钟在向前推进，而不是在后退
 *  Watermark与**数据的时间戳相关**  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 4]

### 2.3 Watermark的传递 ###

多个分区Watermark不同时，取最小值的Watermark，再将新的Watermark广播给下游算子；Watermark不更新时，不用广播  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 5]

### 2.4 Watermark的引入 ###

Event Time的使用一定要指定数据源中的时间戳；  
调用`dataStream.assignTimestampsAndWatermarks`方法，传入一个`BoundedOutOfOrdernessTimestampExtractor`，即可指定Watermark

public class WindowTest4_UDFTimeStampAssigner { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); //默认是处理时间
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            // java8 中的lamda表达式
            DataStream<SensorReading> dataStream = inputStream.map(line -> { 
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            //升序数据设置事件时间和watermark
            dataStream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<SensorReading>() { 
                @Override
                public long extractAscendingTimestamp(SensorReading element) { 
                    return element.getTimestamp() * 1000L;
                }
            });
    
            //乱序数据设置时间戳和watermark
            //BoundedOutOfOrdernessTimestampExtractor 有界乱序时间戳提取器
            dataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) { 
                @Override
                public long extractTimestamp(SensorReading sensorReading) { 
                    return sensorReading.getTimestamp() * 1000L;
                }
            });
    
            env.execute();
        }
    }

**自定义的周期分配器：**

public static class MyPeriodicAssigner implements AssignerWithPeriodicWatermarks<SensorReading>{ 
       private Long bound = 60 * 1000L; //延迟一分钟
       private Long maxTs = Long.MIN_VALUE; //当前最大时间戳
    
       @Nullable
       @Override
       public Watermark getCurrentWatermark() { 
           return new Watermark(maxTs - bound);
       }
    
       @Override
       public long extractTimestamp(SensorReading element, long previousElementTimestamp) { 
          maxTs = Math.max(maxTs, element.getTimestamp());
          return element.getTimestamp();
       }
    }

**自定义的断点分配器：**

//断点分配器
    public static class MyPunctuatedAssigner implements AssignerWithPunctuatedWatermarks<SensorReading>{ 
       private Long bound = 60 * 1000L; //延迟一分钟
    
       @Nullable
       @Override
       public Watermark checkAndGetNextWatermark(SensorReading lastElement, long extractedTimestamp) { 
           if(lastElement.getId().equals("sensor_1")){ 
               return new Watermark(extractedTimestamp - bound);
           }else { 
               return null;
           }
       }
    
       @Override
       public long extractTimestamp(SensorReading element, long previousElementTimestamp) { 
          return element.getTimestamp();
       }
    }

### 2.5 Watermark的设定原则 ###

*  在Flink中，watermark由应用程序开发人员生成，这种通常需要对相应的领域有一定的了解
 *  如果watermark设置的延迟太久，收到结果的速度可能就会很慢，解决办法是在水位线到达之前输出一个近似结果
 *  如果watermark到达太早，则可能使收到错误结果，不过Flink处理迟到数据的机制可以解决这个问题

**事件时间语义下的窗口测试代码1：**

public class WindowTest3_EventTimeWindow { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); //默认是处理时间
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型，分配时间戳和watermark
            // java8 中的lamda表达式
            DataStream<SensorReading> dataStream = inputStream.map(line -> { 
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
         /*   //升序数据设置事件时间和watermark
            dataStream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<SensorReading>() { 
                @Override
                public long extractAscendingTimestamp(SensorReading element) { 
                    return element.getTimestamp() * 1000L;
                }
            });*/
    
            //乱序数据设置时间戳和watermark
            //BoundedOutOfOrdernessTimestampExtractor 有界乱序时间戳提取器
            dataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) { 
                @Override
                public long extractTimestamp(SensorReading sensorReading) { 
                    return sensorReading.getTimestamp() * 1000L;
                }
            });
    
            // 基于事件时间的开窗聚合，统计15秒内温度的最小值
            SingleOutputStreamOperator<SensorReading> minTempStream  = dataStream.keyBy("id")
                    .timeWindow(Time.seconds(15))
                    .minBy("temperature");
    
            minTempStream.print("minTemp");
            env.execute();
        }
    }

**事件时间语义下的窗口测试代码2----迟到数据处理：**

public class WindowTest3_EventTimeWindow { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); //默认是处理时间
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型，分配时间戳和watermark
            // java8 中的lamda表达式
            DataStream<SensorReading> dataStream = inputStream.map(line -> { 
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
         /*   //升序数据设置事件时间和watermark
            dataStream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<SensorReading>() { 
                @Override
                public long extractAscendingTimestamp(SensorReading element) { 
                    return element.getTimestamp() * 1000L;
                }
            });*/
    
            //乱序数据设置时间戳和watermark
            //BoundedOutOfOrdernessTimestampExtractor 有界乱序时间戳提取器
            dataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) { 
                @Override
                public long extractTimestamp(SensorReading sensorReading) { 
                    return sensorReading.getTimestamp() * 1000L;
                }
            });
    
            OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("late") { 
            };
            // 基于事件时间的开窗聚合，统计15秒内温度的最小值
            SingleOutputStreamOperator<SensorReading> minTempStream  = dataStream.keyBy("id")
                    .timeWindow(Time.seconds(15))
                    .allowedLateness(Time.minutes(1))
                    .sideOutputLateData(outputTag)
                    .minBy("temperature");
    
            minTempStream.print("minTemp");
            minTempStream.getSideOutput(outputTag).print("late");
            env.execute();
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70]: /images/20221022/1e2ec25bc19347398af3a7d0649dab6e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 1]: /images/20221022/50c51dc44ba44e3cb925638fef53ac50.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 2]: /images/20221022/cb1538f601f84a4f929c464d0195eecd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 3]: /images/20221022/97f60cf4dd314f3ca595497ac0a2d6dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 4]: /images/20221022/efba52a460f34fbd8ca32dcd7a5418d9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 5]: /images/20221022/887f81b00ebb489499b438926ed577e5.png