西部数码做的网站打不开,python如何与wordpress,医疗网站设计网站,做排版的网站Flink 中的时间语义 时间语义#xff1a; EventTime#xff1a;事件创建时间#xff1b;Ingestion Time#xff1a;数据进入Flink的时间#xff1b;Processing Time#xff1a;执行操作算子的本地系统时间#xff0c;与机器无关。不同的时间语义有不同的应用场合#x…Flink 中的时间语义 时间语义 EventTime事件创建时间Ingestion Time数据进入Flink的时间Processing Time执行操作算子的本地系统时间与机器无关。不同的时间语义有不同的应用场合我们往往更关系事件时间Event Time。数据生成的时候就会自动注入时间戳Event Time可以从日志数据的时间戳timestamp)中提取。 设置 Event Time 我们可以直接在代码中对执行环境调用setStreamTimeCharacteristic方法设置流的时间特性。具体的时间还需要从数据中提取时间戳timestamp。 val env StreamExecutionEnvironment.getExecutionEnvironment //从调用时刻开始给 env 创建的每一个 stream 追加时间特性 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)乱序数据的影响 当Flink以Event Time模式处理数据流时它会根据数据里的时间戳来处理基于时间的算子。由于网络、分布式等原因会导致乱序数据的产生。如上图所示理想情况与实际情况会存在差异乱序数据会让窗口计算不准确。解决方案是让窗口等几分钟。 水位线 Watermark 怎么避免乱序数据带来计算不正确 遇到一个时间戳到达了窗口关闭时间不应该立刻触发窗口计算而是等待一段时间等迟到的数据来了再关闭窗口。Watermark是一种衡量Event Time进展的机制可以设置延迟触发。Watermark是用于处理乱序事件的而正确的处理乱序事件通常用Watermark机制结合window来实现。数据流中的Watermark用于表示timestamp小于Watermark的数据都已经达到了因此window的执行也是由Watermark触发的。Watermark用来让程序自己延迟和结果正确性。 Watermark 的特点 Watermark是一条特殊的数据记录必须单调递增以确保任务的事件时间时钟在向前推进而不是在后退。Watermark与数据的时间戳有关。 watermark 的传递、引入和设定 watermark的传递 一个Task输入可以并行多个如下有4个并行度输出也可能存在多个并行如下有3个。每个任务Task内部都有一个事件时钟且每个分区也维护了对应的WM如下的Partition WM。当事件流流进Partition时会判断新事件流的WM是否大于当前的Partition WM当大于时就更新Partition的时间戳WM为新流入的WM取最大值如下1-2象限Partition WM的变化。同时如下Task也维护了一个全局的WM表示事件时钟该值取分区中最小的WM作为输出的时间戳如下第二象限的输出选择最小的WM3向下传递。当第二个(横线)分区Partition WM流进来WM7的事件流时就会出现第三象限的情景但是最小的WM还是3因此不更新Task全局的WM。当第三个分区Partition WM流进来WM6的事件流时就会出现第四象限的情景此时分区Partition WM的最小值4因此Task全局WM就4。 watermark的引入 Event Time的使用一定要指定数据源中的时间戳。对于排好序的数据只需要指定时间戳就够了不需要延迟触发。 import org.apache.flink.streaming.api.windowing.time.Time //同时分配时间戳和水位线 dataStream.assignTimestampsAndWatermarks( //无序数据 Time.milliseconds(1000)延迟时间 new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.milliseconds(1000)) {//提取事件戳 timestamp * 1000是因为出入的毫秒override def extractTimestamp(t: SensorReading): Long {t.timestamp * 1000} })【1】对于排好序的数据不需要延迟触发可以只指定事件戳就行了 dataStream.assignTimestampsAndWatermarks(_.timestamp * 1000)【2】Flink暴露了TimestampAssigner接口供我们实现使我们可以自定义如何从事件数据中抽取时间戳和生成 watermark。MyAssigner可以有两种类型都继承自TimestampAssigner。 dataStream.assignTimestampsAndWatermarks(new MyAssigner())TimestampAssigner定义了抽取时间戳以及生成watermark的方法有两种类型 【1】AssignerWithPeriodicWatermarks 系统会周期性的将Watermark插入到流中。默认周期是200毫秒如果是processingTime则Watermark 0 可以使用ExecutionConfig.setAutoWatermarkInterval()方法进行设置。升序和前面乱序的处理BoundedOutOfOrderness都是基于周期性watermark的。举例如下产生watermark的逻辑每隔5秒Flink调用AssignerWithPeriodicWatermarks的getCurrentWatermark()方法。如果方法返回一个时间戳大于之前水位的时间戳新的water会被插入到流中。这个检查保证了水位线是单调递增的。如果方法返回的时间戳小于之前水位的时间戳则不会产生新的watermark。 //方案一: //EventTime是以数据自带的时间戳字段为准应用程序需要指定如何从record中抽取时间戳字段 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //每隔 5秒产生一个 watermark env.getConfig.setAutoWatermarkInterval(5000);​ //方案二 //自定义一个周期性的时间戳 class PeriodicAssigner extends AssignerWithPeriodicWatermarks[SensorReading]{val bound: Long 60 * 1000 //延时为 1 分钟var maxTs: Long Long.MinValue //观察到的最大时间戳//生成水位线override def getCurrentWatermark: Watermark {new Watermark(maxTs - bound)}//抽取时间戳的方法override def extractTimestamp(t: SensorReading, l: Long): Long {maxTs maxTs.max(t.timestamp)t.timestamp} }【2】AssignerWithPunctuatedWatermarks 没有时间周期规律可打断的生成watermark。 class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[SensorReading]{val bound: Long 60 * 1000//获取水位线根据数据触发override def checkAndGetNextWatermark(t: SensorReading, l: Long): Watermark {if(t.id sensor_1){new Watermark(l - bound)}else{null}}//抽取时间戳的方法override def extractTimestamp(t: SensorReading, l: Long): Long {t.timestamp} }watermark 的设定 【1】在Flink中watermark由应用程序开发人员生成这通常需要对相应的领域有一定的了解。 【2】如果watermark设置的延迟太久收到结果的速度可能就会很慢解决办法是在水位线到达之前输出一个近似结果。 【3】而如果watermark到达得太早则可能收到错误结果不过Flink处理迟到数据的机制可以解决这个问题。