详解 Flink 的时间语义和 watermark

2024-06-15 1663阅读

一、Flink 时间语义类型

详解 Flink 的时间语义和 watermark

  • Event Time：是事件创建的时间。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink 通过时间戳分配器访问事件时间戳
  • Ingestion Time ：是数据进入 Flink 的时间
  • Processing Time：是每一个执行基于时间操作的算子的本地系统时间，与机器相关，默认的时间属性就是 Processing Time

    二、EventTime 引入

    Flink 默认是按照 ProcessingTime 来处理数据的

    /**
	在 Flink 的流式处理中，绝大部分情况推荐使用 eventTime，一般只在 eventTime 无法使用时，才会被迫使用 ProcessingTime 或者 Ing estionTime 。使用 EventTime ，需要先引入 EventTime 的时间属性
*/
public class EventTimeTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        //引入 EvenetTime
        //TimeCharacteristic 是一个枚举类，有 ProcessingTime、IngestionTime 和 EventTime 三个属性
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        
    }
}

    三、Watermark

    1. 数据乱序情况

    详解 Flink 的时间语义和 watermark

    • 正常情况下，Flink 接收到的事件应该要是按照事件的产生时间 (EventTime) 的先后顺序排列的
    • 实际情况下，事件从产生到进入 source 再到触发 operator，其中间是有一个过程和时间的，而且由于网络、分布式等原因会造成 Flink 接收到的事件的先后顺序不是严格按照事件的 EventTime 顺序排列的，即所谓的乱序数据
    • 乱序数据的问题会造成窗口触发关闭的时间混乱，计算不准确
    • Flink 处理乱序数据的机制：Watermark + allowedLateness + sideOutputLateData

      2. Watermark 介绍

      • Watermark 是一种使用延迟触发 window 执行来处理乱序数据的机制
      • 原理：当设置 Watermark = t 时 (即延迟时长为 t)，则 Flink 每一次都会获取已经到达的数据中的最大的 EventTime，然后判断 maxEventTime - t 是否等于某一个窗口的触发时间，如果相等则认为属于这个窗口的所有数据都已经到达，这个窗口被触发执行关闭，也可能存在数据丢失
      • 在数据有序的流中，相当于 Watermark = 0，即已经到达的数据中的最大的 EventTime 等于某一个窗口的触发时间，则这个窗口被触发执行关闭
      • 一般将 Watermark 设置为乱序数据流中最大的迟到时间差

        3. Watermark 特点和行为

        • 水位线 (Watermark) 是作为一个特殊的数据插入到数据流中的一个标记
        • 水位线 (Watermark) 在 Flink 程序中是一个常量类，有一个时间戳属性，用来表示当前事件时间的进展
        • 水位线 (Watermark) 是基于数据的 EventTime 时间戳生成的
        • 水位线 (Watermark) 的时间戳必须单调递增，以确保任务的事件时间时钟一直向前推进

          4. Watermark 在任务间的传递

          任务并行度不为 1；Watermark 设置的位置越靠近 Source 端越好

          详解 Flink 的时间语义和 watermark

          • 一个任务会接收上游多个并行任务的数据，也会向下游多个并行任务发送数据
          • 从上游多个并行任务接收 Watermark：使用 Partition WM 分别存储接收到的不同分区任务的 Watermark，并以其中最小的 Watermark 作为自己当前的事件时间
          • 向下游多个并行任务发送 Watermark：采取广播的分区策略，向下游的每一个任务都发送一份 Watermark，如果后续 Watermark 没有变更则不会重复发送

            5. Watermark 引入

            5.1 核心代码
            /**
	方法签名：
		DataStream.assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks)
		DataStream.assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks)
	
	参数：
		1.AssignerWithPeriodicWatermarks：继承 TimestampAssigner 接口，周期性的生成 watermark，常用实现类为：BoundedOutOfOrdernessTimestampExtractor 和 AscendingTimestampExtractor
		2.AssignerWithPunctuatedWatermarks：继承 TimestampAssigner 接口，间断式地生成 watermark
*/
public class WatermarkTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        //引入 EvenetTime       
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        
        DataStream dataStream = env.socketTextStream("localhost", 7777);
        
        DataStream inputStream = dataStream.map(new MapFunction() {
            @Override
            public SensorReading map(String value) throws Exception {
                String[] fields = value.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            }
        });
        
        //有序数据设置事件时间戳(毫秒数)和watermark
        //不需要传递watermark延迟时间，默认是当前事件时间戳 - 1ms 作为watermark
        inputStream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor() {
            @Override
            public long extractAscendingTimestamp(SensorReading element) {
                return element.getTimestamp() * 1000L;
            }
        });
        
        //乱序数据设置事件时间戳(毫秒数)和watermark
        //BoundedOutOfOrdernessTimestampExtractor 构造方法必须传入watermark延迟时间
        //生成的watermark时间戳 = 当前所有事件的最大时间戳 - 延迟时间
        inputStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor(Time.seconds(2)) {
            @Override
            public long extractTimestamp(SensorReading element) {
                return element.getTimestamp() * 1000L;
            }
        });
        
        env.execute();
        
    }
}
            5.2 AssignerWithPeriodicWatermarks

            系统会周期性地生成 watermark 并插入到数据流中，默认周期是 200 毫秒

            /**
	设置watermark生成周期：env.getConfig.setAutoWatermarkInterval(milliseconds);
	产生watermark的逻辑：每隔 0.2 秒钟，Flink 会调用 AssignerWithPeriodicWatermarks 的 getCurrentWatermark() 方法获取一个时间戳，如果大于之前水位的时间戳，新的 watermark 会被插入到流中。这个检查保证了水位线是单调递增的。如果方法返回的时间戳小于等于之前水位的时间戳，则不会产生新的 watermark
	自定义watermark周期生成器：实现 AssignerWithPeriodicWatermarks 接口，并重写 getCurrentWatermark 和 extractTimestamp 方法
*/
public class MyPeriodicAssigner implements AssignerWithPeriodicWatermarks {
    private Long bound = 60 * 1000L;  // watermark延迟时间
    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();
    }
}
            5.3 AssignerWithPunctuatedWatermarks

            间断式地生成 watermark，可以根据需要对每条数据进行条件判断筛选来确定是否生成 watermark

            public class MyPunctuatedAssigner implements AssignerWithPunctuatedWatermarks {
    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();
    }
}

            四、EventTime 的 window 操作

            1. 滚动时间窗口操作

            /**
	需求：统计 15 秒内的最小温度值，设置 2 秒的延迟
*/
public class TumblingEventTimeWindowTest {
 	public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        
        /*
          sensor_1,1547718199,35.8
          sensor_6,1547718201,15.4
          sensor_7,1547718202,6.7
          sensor_10,1547718205,38.1
          sensor_1,1547718207,36.3
          sensor_1,1547718209,32.8
          sensor_1,1547718212,37.1
          ...
        */
        DataStream inputStream = env.socketTextStream("localhost", 7777);
        
        DataStream dataStream = inputStream.map(new MapFunction() {
            @Override
            public SensorReading map(String value) {
                String[] fields = value.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            }
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor(Time.seconds(2)) {
            @Override
            public long extractTimestamp(SensorReading element) {
                return element.getTimestamp() * 1000L;
            }
        });
        
        //开窗聚合
       SingleOutputStreamOperator minTempStream = dataStream.keyBy("id").timeWindow(Time.seconds(15)).minBy("temperature");
        
        minTempStream.print("minTemp");
        
        /**
        	输出的结果分析：
        		1.在接收到 sensor_1,1547718212,37.1 时，触发了一个窗口关闭，此时数据的 EventTime 为 1547718212，由于 watermark 延迟时间设置为 2，所以该窗口触发关闭的时间戳为 1547718212 - 2 = 1547718210，该窗口的范围为 [1547718195,1547718210)
        		2.当前第一个窗口是 [1547718195,1547718210)，其起始点的确定规则为：
        			2.1 滚动时间窗口使用的窗口分配器为 TumblingEventTimeWindows 类
        			2.2 TumblingEventTimeWindows 的 assignWindows 方法中调用 getWindowStartWithOffset 方法获取起始点
        			2.3 getWindowStartWithOffset(timestamp, offset, windowSize)：方法逻辑为 timestamp - (timestamp - offset + windowSize) % windowSize，默认 offset 为 0，所以最终得到的起始点应该是 windowSize 的整数倍，在本例中的起始点为 1547718199 - (1547718199-0+15)%15 = 1547718195
        		3.偏移量 offset：一般是用来处理不同时区的数据
        */
        
        env.execute();
        
    }   
}

            2. 迟到数据处理

            /**
	需求：统计 15 秒内的最小温度值，设置 2 秒的延迟，并允许 1 分钟的迟到数据，1 分钟后的数据写入侧输出流
*/
public class TumblingEventTimeWindowDelayTest {
 	public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        
        DataStream inputStream = env.socketTextStream("localhost", 7777);
        
        DataStream dataStream = inputStream.map(new MapFunction() {
            @Override
            public SensorReading map(String value) {
                String[] fields = value.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            }
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor(Time.seconds(2)) {
            @Override
            public long extractTimestamp(SensorReading element) {
                return element.getTimestamp() * 1000L;
            }
        });
        
        OutputTag outputTag = new OutputTag("late"){};
        
        //开窗聚合
       SingleOutputStreamOperator minTempStream = dataStream.keyBy("id")
           .timeWindow(Time.seconds(15))
           .allowedLateness(Time.minutes(1));
           .sideOutputLateData(outputTag)
           .minBy("temperature");
        
        minTempStream.print("minTemp");
        minTempStream.getSideOutput(outputTag).print("late");
        
        /**
        	依次输入数据：
              sensor_1,1547718199,35.8
              sensor_1,1547718206,36.3
              sensor_1,1547718210,34.7
              sensor_1,1547718211,31
              sensor_1,1547718209,34.9
              sensor_1,1547718212,37.1
              sensor_1,1547718213,33
              sensor_1,1547718206,34.2
              sensor_1,1547718202,36
              ...
              sensor_1,1547718272,34
              sensor_1,1547718203,30.6
        
        	输出的结果分析：
        		1.在接收到 sensor_1,1547718212,37.1 时，触发 [1547718195,1547718210) 窗口执行，此时输出数据 sensor_1,1547718209,34.9，此时 2 秒内的延迟数据能被处理  
        		2.在接收到 sensor_1,1547718206,34.2 时，由于设置了允许 1 分钟迟到，所以 [1547718195,1547718210) 窗口仍然没有关闭，此时会更新数据为 sensor_1,1547718206,34.2，此时的系统时间戳为 1547718213 - 2 = 1547718211 - 1547718210 = 60，所以 [1547718195,1547718210) 窗口会真正的关闭
        		5.在之后接收到 sensor_1,1547718203,30.6 时，会把数据输出到侧输出流中
        */
        
        env.execute();
        
    }   
}
VPS购买请点击我

免责声明：我们致力于保护作者版权，注重分享，被刊用文章因无法核实真实出处，未能及时与作者取得联系，或有版权异议的，请联系管理员，我们会立即处理! 部分文章是来自自研大数据AI进行生成,内容摘自(百度百科,百度知道,头条百科,中国民法典,刑法,牛津词典,新华词典,汉语词典,国家院校,科普平台)等数据,内容仅供学习参考,不准确地方联系删除处理! 图片声明：本站部分配图来自人工智能系统AI生成,觅知网授权图片,PxHere摄影无版权图库和百度，360，搜狗等多加搜索引擎自动关键词搜索配图，如有侵权的图片，请第一时间联系我们，邮箱：ciyunidc@ciyunshuju.com。本站只作为美观性配图使用,无任何非法侵犯第三方意图,一切解释权归图片著作权方,本站不承担任何责任。如有恶意碰瓷者,必当奉陪到底严惩不贷!

相关阅读

目录[+]