本篇内容主要讲解“flink1.2版本时间、水位线的介绍和用法”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“flink1.2版本时间、水位线的介绍和用法”吧!
水位线是flink的一种处理延时数据的机制，主要对设定时间内延时数据的自动容错，水位线的本质是时间戳，计算公式为：当前事件最大时间值 - 数据延时时间。（看了几遍有点懵）
个人理解：
水位线是收到数据逻辑时间便签，是处理延时数据的基础，通过与数据自带的生成时间Timestamps，实现延迟数据矫正。
理想状态下的水位线，即数据元素的事件事件是有序的，Watermark时间戳会随着数据元素的事件时间安装顺序生成，此时，水位线时间和时间时间保持一致。
现实情况数据元素往往并不按照其生产顺序接入Flink，而频繁处理乱序或迟到情况，这时候需要watermark来处理，当事件8和事件11同时进入系统，flink系统将根据设定延时值分别计算它们的watermark，两个事件到达一个operator中后，匹配事件时间的虚拟时间与watermark匹配，触发响应的计算。
Watermark在Source Operator中生成，且在每个Operator的子Task中独立生成。
如果一个watermark同时更新一个算子Task的当前事件时间，Flink会选择最小的水位线进行更新。当一个Window算子Task中水位线大于Window结束时间，立即触发窗口计算。
流式处理中最大的特点是数据上具有时间的属性特征，Flink根据时间产生的位置不同，将时间分为三种概念：事件生成时间（Event Time）、事件接入时间（Ingestion TIme）、事件处理时间（Processing Time）。
事件生成时间：数据从终端或系统中产生的过程消耗的时间。
数据接入时间：数据接入DataSource时的时间。
事件处理时间：处理过程中获取的主机时间。
Timestamps和Watermark成对对存在，使用时，都要指定
watermark设定Flink中Watermark默认200ms生成一次，也可以手动指定，代码如下：
// 1、创建flink运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(3); // 设置并行度 env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC); //处理模式设定：流或批 // 生成 watermark 的时间间隔（每 n 毫秒）,设置周期性的产生水位线的时间间隔。当数据流很大的时候，如果每个事件都产生水位线，会影响性能。 //env.getConfig().setAutoWatermarkInterval(1000); // 自动水印时间间隔 12版本不用设置，有默认
此处以滚动窗口为例，窗口知识下次分享，首先对数据进行机构化，数据结构："yyyy-MM-dd HH:mm:ss|type|num"，处理代码如下：
SingleOutputStreamOperator<Tuple3<String,String, Integer>> formatData =text.map(new MapFunction<String, Tuple3<String, String, Integer>>() {
// 数据格式转换
private static final long serialVersionUID = 1L;
@Override
public Tuple3<String, String, Integer> map(String value) throws Exception {
Tuple3<String, String, Integer> data = new Tuple3<String, String, Integer>();
String[] dataTmp = value.split("\\|");
data.f0 = dataTmp[0];
data.f1 = dataTmp[1];
data.f2 = Integer.parseInt(dataTmp[2]);
return data;
}
});设置Timestamps和最大时延
SingleOutputStreamOperator<Tuple3<String,String, Integer>> orderDSWithWatemark=formatData .assignTimestampsAndWatermarks( // 设置watermark watemark = 最大事件时间 - 最大延迟或乱序时间 WatermarkStrategy.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(3)) //指定maxOutOfOrderness最大无序度时间即最大延迟时间/乱序时间 .withTimestampAssigner((data,timestamp) -> Long.parseLong(DateUtil.dateToUTC(data.f0))*1000) //时间为毫秒级 );
设定窗口大小和处理逻辑
SingleOutputStreamOperator<Tuple3<String,String, Integer>> result=orderDSWithWatemark.keyBy(one -> one.f1)
.window(TumblingEventTimeWindows.of(Time.seconds(10))) // 设定窗口大小
// .allowedLateness(Time.seconds(1)) //延时处理时间
// .sideOutputLateData(lateOutputTag) //侧输出
.reduce(new ReduceFunction<Tuple3<String, String, Integer>>() { // 处理逻辑
private static final long serialVersionUID = -6695049408336015245L;
@Override
public Tuple3<String, String, Integer> reduce(Tuple3<String, String, Integer> value1,
Tuple3<String, String, Integer> value2) throws Exception {
Tuple3<String, String, Integer> data = new Tuple3<String, String, Integer>();
data.f0 = value2.f0;
data.f1 = value1.f1;
data.f2 = value1.f2 + value2.f2;
System.out.println(data);
return data;
}
});
result.print("滚动事件时间");
env.execute();时间和水位线是flink中比较难理解且重要的概念，我也是一知半解，在使用的过程中再慢慢深化，基本逻辑是针对数据建立自己的时间标签，并通过时间范围（窗口）和数据延迟完成事件内数据的汇集、计算和输出，以此，完成更精确的实时事件数据计算。
技术是需求的一种呈现，基础本质相互交叠，编程语言、技术框架都是，最重要的细微处的优化和整体的使用的简便，功能的稳定和强大。
到此，相信大家对“flink1.2版本时间、水位线的介绍和用法”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
