义乌做网站要多少钱,备案密码如何登录添加网站,宝丰网站建设,如何用云服务器做网站一、Flink 中的水印
在Flink的流式操作中, 会涉及不同的时间概念#xff1a;
1.1 处理时间
是指的执行操作的各个设备的时间#xff0c;对于运行在处理时间上的流程序, 所有的基于时间的操作(比如时间窗口)都是使用的设备时钟。比如, 一个长度为1个小时的窗口将会包含设备…一、Flink 中的水印
在Flink的流式操作中, 会涉及不同的时间概念
1.1 处理时间
是指的执行操作的各个设备的时间对于运行在处理时间上的流程序, 所有的基于时间的操作(比如时间窗口)都是使用的设备时钟。比如, 一个长度为1个小时的窗口将会包含设备时钟表示的1个小时内所有的数据。 假设应用程序在 9:15am分启动, 第1个小时窗口将会包含9:15am到10:00am所有的数据然后下个窗口是10:00am-11:00am, 等等。处理时间是最简单时间语义, 数据流和设备之间不需要做任何的协调。他提供了最好的性能和最低的延迟。 但是, 在分布式和异步的环境下处理时间没有办法保证确定性容易受到数据传递速度的影响: 事件的延迟和乱序。在使用窗口的时候, 如果使用处理时间, 就指定时间分配器为处理时间分配器。
1.2 事件时间
是指的这个事件发生的时间。在event进入Flink之前, 通常被嵌入到了event中 一般作为这个event的时间戳存在。在事件时间体系中, 时间的进度依赖于数据本身和任何设备的时间无关。事件时间程序必须制定如何产生Event Time Watermarks(水印) 。假设所有数据都已到达事件时间操作将按预期方式运行即使在处理无序或迟到的事件或重新处理历史数据时也会产生正确且一致的结果。例如每小时事件时间窗口将包含带有事件时间戳的所有记录这些记录落入该小时。在使用窗口的时候, 如果使用事件时间, 就指定时间分配器为事件时间分配器。从1.12开始, Flink内部已经把默认的语义改成了事件时间。
1.3 Flink中的WaterMark
支持event time的流式处理框架需要一种能够测量event time 进度的方式。 比如, 一个窗口算子创建了一个长度为1小时的窗口,那么这个算子需要知道事件时间已经到达了这个窗口的关闭时间从而在程序中去关闭这个窗口。事件时间可以不依赖处理时间来表示时间的进度。例如在程序中 即使处理时间和事件时间有相同的速度 事件时间可能会轻微的落后处理时间。另外一方面使用事件时间可以在几秒内处理已经缓存在Kafka中多周的数据这些数据可以照样被正确处理, 就像实时发生的一样能够进入正确的窗口。这种在Flink中去测量事件时间的进度的机制就是watermark(水印)。
1.4 Flink中如何产生水印 二、代码集成
package com.lyh.flink08;import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;import javax.naming.Context;
import java.time.Duration;public class WatorMark_01 {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperatorWaterSensor stream env.socketTextStream(hadoop100, 9999).map(new MapFunctionString, WaterSensor() {Overridepublic WaterSensor map(String value) throws Exception {String[] datas value.split(,);return new WaterSensor(datas[0],Long.valueOf(datas[1]),Integer.valueOf(datas[2]));}});WatermarkStrategyWaterSensor wms WatermarkStrategy.WaterSensorforBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssignerWaterSensor() { // 指定时间戳Overridepublic long extractTimestamp(WaterSensor element, long recordTimestamp) {return element.getTs() * 1000;}});stream.assignTimestampsAndWatermarks(wms) // 指定水印和时间戳.keyBy(WaterSensor::getId).window(TumblingEventTimeWindows.of(Time.seconds(5))).process(new ProcessWindowFunctionWaterSensor, String, String, TimeWindow() {Overridepublic void process(String key,Context ctx,IterableWaterSensor elements,CollectorString out) throws Exception {String msg 当前key: key 窗口: [ ctx.window().getStart() / 1000 , ctx.window().getEnd()/1000 ) 一共有 elements.spliterator().estimateSize() 条数据 ;out.collect(msg);}}).print();env.execute();}
}三、测试结果
