网络工程毕业后干什么,拱墅抖音seo搜索排名推广,做全屏网站设计时容易犯的错,wordpress 来源统计一、Time 在Flink的流式处理中#xff0c;会涉及到时间的不同概念 Event Time#xff1a;是事件创建的时间。它通常由事件中的时间戳描述#xff0c;例如采集的日志数据中#xff0c;每一条日志都会记录自己的生成时间#xff0c;Flink通过时间戳分配器访问事件时间戳 Ing…一、Time 在Flink的流式处理中会涉及到时间的不同概念 Event Time是事件创建的时间。它通常由事件中的时间戳描述例如采集的日志数据中每一条日志都会记录自己的生成时间Flink通过时间戳分配器访问事件时间戳 Ingestion Time是数据进入Flink的时间 Processing Time是每一个执行基于时间操作的算子的本地系统时间与机器相关默认的时间属性就是Processing Time。 例如一条日志进入Flink的时间为2017-11-12 10:00:00.123 到达window的系统时间为 2017-11-12 10:00:01.234日志内容如下 2017-11-02 18:37:15.624 INFO Fair over to rm2 对于业务来说要统计1min内的故障日志个数哪个时间是最有意义的----- eventTime因为我们要根据日志的生成时间进行统计。    如果要想聚合不可能对无解数据流进行聚合。   二、Window 1、streaming流式计算是一种被设计用于处理处理无限数据集的数据处理引擎而无限数据集是指一种不断增长的本质上无限的数据集而window是一种切割无限数据为有限块进行处理的手段。 Window是无限数据流处理的核心Window将一个无限的stream拆分成有限大小的buckets桶我们可以在这些桶上做计算操作。 共有两类五种时间窗口。 2、Window类型两类 2.1、CountWindow按照指定的数据条数生成一个window与时间无关 2.2、TimeWindow按照时间生成window。按照Processing Time来划分Window 对于TimeWindow和CountWindow可以根据窗口实现原理的不同分成三类滚动窗口Tumbling Window、滑动窗口Sliding Window和会话窗口Session Window。 1滚动窗口Tumbling Windows 将数据依据固定的窗口长度对数据进行切分。 特点时间对齐窗口长度固定没有重叠。 滚动窗口分配器将每个元素分配到一个指定窗口大小的窗口中滚动窗口有一个固定的大小并且不会出现重叠。 2滑动窗口Sliding Windows 滑动窗口是固定窗口的更广义的一种形式滑动窗口由固定的窗口长度和滑动间隔组成。 特点时间对齐窗口长度固定有重叠。 滑动窗口分配器将元素分配到固定长度的窗口中与滚动窗口类似窗口的大小由窗口大小参数来配置另一个窗口滑动参数控制滑动窗口开始的频率。 因此滑动窗口如果滑动参数小于窗口大小的话窗口是可以重叠的在这种情况下元素会被分配到多个窗口中。 使用场景对最近一个时间段内的统计求某接口最近5min的失败率来决定是否要报警。 3会话窗口Session Windows 由一系列事件组合一个指定时间长度的timeout间隙组成。类似于web应用的session也就是一段时间没有接收到新数据就会生成新的窗口。 特点时间无对齐。 session 窗口分配器通过session活动来对元素进行分组session窗口跟滚动窗口和滑动窗口相比不会有重叠和固定的开始时间和结束时间的情况相反当它在一个固定的 时间周期内不再收到元素即非活动间隔产生那这个窗口就会关闭。一个Session窗口通过一个session间隔来配置这个session间隔定义了非活跃周期的长度当这个非活跃 周期产生那么当前的session将关闭并且后续的元素将被分配到新的session窗口中去。   三、Window API 3.1、CountWindow CountWindow根据窗口中相同key元素的数量来触发执行执行时只计算元素数量达到窗口大小的key对应的结果。 注意CountWindow的window_size 指的是相同key的元素的个数不是输入的所有元素的总数。 import org.apache.flink.api.java.tuple.Tuple import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}/*** CountWindow 中的滚动窗口Tumbling Windows* 将数据依据固定的窗口长度对数据进行切分。*/ object TimeAndWindow {def main(args: Array[String]): Unit {val env StreamExecutionEnvironment.getExecutionEnvironmentval stream: DataStream[String] env.socketTextStream(localhost,11111)val streamKeyBy: KeyedStream[(String, Long), Tuple] stream.map(item (item,1L)).keyBy(0)//注意CountWindow的window_size 指的是相同key的元素的个数不是输入的所有元素的总数。val streamWindow: DataStream[(String, Long)] streamKeyBy.countWindow(5).reduce((item1, item2)(item1._1,item1._2item2._2))streamWindow.print()env.execute(TimeAndWindow)} } 3.2 import org.apache.flink.api.java.tuple.Tuple import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}/*** CountWindow 中的滑动窗口Sliding Windows* 将数据依据固定的窗口长度对数据进行切分。*/ object TimeAndWindow {def main(args: Array[String]): Unit {val env StreamExecutionEnvironment.getExecutionEnvironmentval stream: DataStream[String] env.socketTextStream(localhost,11111)val streamKeyBy: KeyedStream[(String, Long), Tuple] stream.map(item (item,1L)).keyBy(0)//注意CountWindow的window_size 指的是相同key的元素的个数不是输入的所有元素的总数。//满足步长就执行一次按第一个参数的长度val streamWindow: DataStream[(String, Long)] streamKeyBy.countWindow(5,2).reduce((item1, item2)(item1._1,item1._2item2._2))streamWindow.print()env.execute(TimeAndWindow)} } 四、EventTime与Window 1、EventTime的引入 在Flink的流式处理中绝大部分的业务都会使用eventTime一般只在eventTime无法使用时才会被迫使用ProcessingTime或者IngestionTime。 如果要使用EventTime那么需要引入EventTime的时间戳引入方式如下所示 2、Watermark 　　概念我们知道流处理从事件产生到流经source再到operator中间是有一个过程和时间的虽然大部分情况下流到operator的数据都是按照事件产生的 事件戳顺序来的但是也不排除由于网络、背压等原因导致乱序的产生所谓乱序就是指Flink接收到的事件的先后顺序不是严格按照事件的EventTime顺序排列的。 　　Watermark是一种衡量Event Time进展的机制它是数据本身的一个隐藏属性数据本身携带着对应的Watermark。 　　Watermark是用于处理乱序事件的而正确的处理乱序事件通常用Watermark机制结合window来实现。 　　数据流中的Watermark用于表示eventTime小于Watermark的数量都已经到达了因此window的执行也是由Watermark触发的。 　　Watermark可以理解成一个延迟触发机制。我们可以设置Watermark的延时时长t每次系统会校验已经到达的数据中最大的maxEventTime然后认定eventTime 小于 maxEventTime-t 的所有数据都已经到达。如果有窗口的停止时间等于maxEventTime-t那么这个窗口被触发执行。 滚动窗口/滑动窗口/会话窗口 import org.apache.flink.api.java.tuple.Tupleimport org.apache.flink.streaming.api.TimeCharacteristicimport org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractorimport org.apache.flink.streaming.api.scala._import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, SlidingEventTimeWindows, TumblingEventTimeWindows}import org.apache.flink.streaming.api.windowing.time.Timeimport org.apache.flink.streaming.api.windowing.windows.TimeWindow/** * TimeWindow */object EventTimeAndWindow { def main(args: Array[String]): Unit { val env StreamExecutionEnvironment.getExecutionEnvironment //开启watermark //从调用时刻开始给env创建的每一个stream追加时间特征。 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) val stream: KeyedStream[(String, Long), Tuple] env.socketTextStream(192.168.218.130, 1111).assignTimestampsAndWatermarks( new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(3000)) { override def extractTimestamp(element: String): Long { // event word eventTime是日志生成时间我们从日志中解析EventTime val eventTime element.split( )(0).toLong println(eventTime) eventTime } } ).map(item (item.split( )(1),1L)).keyBy(0) //加上滚动窗口窗口大小是5s调用window的api// val streamWindow: WindowedStream[(String, Long), Tuple, TimeWindow] stream.window(TumblingEventTimeWindows.of(Time.seconds(5))) //滑动窗口// val streamWindow: WindowedStream[(String, Long), Tuple, TimeWindow] stream.window(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5))) //会话窗口 val streamWindow: WindowedStream[(String, Long), Tuple, TimeWindow] stream.window(EventTimeSessionWindows.withGap(Time.seconds(5))) val streamReduce streamWindow.reduce((item1,item2)(item1._1,item1._2item2._2)) streamReduce.print() env.execute(EventTimeAndWindow) }}  转载于:https://www.cnblogs.com/ssqq5200936/p/11014296.html