flink1.2版本时间、水位线的介绍和用法

本篇内容主要讲解"flink1.2版本时间、水位线的介绍和用法"，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习"flink1.2版本时间、水位线的介绍和用法"吧!

水位线

• 水位线是flink的一种处理延时数据的机制，主要对设定时间内延时数据的自动容错，水位线的本质是时间戳，计算公式为：当前事件最大时间值 - 数据延时时间。（看了几遍有点懵）

• 个人理解：

• 水位线是收到数据逻辑时间便签，是处理延时数据的基础，通过与数据自带的生成时间Timestamps，实现延迟数据矫正。

种类

顺序事件中的Watermarks

• 理想状态下的水位线，即数据元素的事件事件是有序的，Watermark时间戳会随着数据元素的事件时间安装顺序生成，此时，水位线时间和时间时间保持一致。

乱序事件中的Watermarks

• 现实情况数据元素往往并不按照其生产顺序接入Flink，而频繁处理乱序或迟到情况，这时候需要watermark来处理，当事件8和事件11同时进入系统，flink系统将根据设定延时值分别计算它们的watermark，两个事件到达一个operator中后，匹配事件时间的虚拟时间与watermark匹配，触发响应的计算。

并行数据流中的Watermarks

• Watermark在Source Operator中生成，且在每个Operator的子Task中独立生成。

• 如果一个watermark同时更新一个算子Task的当前事件时间，Flink会选择最小的水位线进行更新。当一个Window算子Task中水位线大于Window结束时间，立即触发窗口计算。

时间概念

• 流式处理中最大的特点是数据上具有时间的属性特征，Flink根据时间产生的位置不同，将时间分为三种概念：事件生成时间（Event Time）、事件接入时间（Ingestion TIme）、事件处理时间（Processing Time）。

• 事件生成时间：数据从终端或系统中产生的过程消耗的时间。

• 数据接入时间：数据接入DataSource时的时间。

• 事件处理时间：处理过程中获取的主机时间。

Event Time

• Timestamps和Watermark成对对存在，使用时，都要指定

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版本不用设置，有默认

指定Timestamps

• 此处以滚动窗口为例，窗口知识下次分享，首先对数据进行机构化，数据结构："yyyy-MM-dd HH:mm:ss|type|num"，处理代码如下：

SingleOutputStreamOperator> formatData =text.map(new MapFunction>() {   // 数据格式转换  private static final long serialVersionUID = 1L;  @Override  public Tuple3 map(String value) throws Exception {    Tuple3 data = new Tuple3();    String[] dataTmp = value.split("\\|");    data.f0 = dataTmp[0];    data.f1 = dataTmp[1];    data.f2 = Integer.parseInt(dataTmp[2]);    return data;  }});

• 设置Timestamps和最大时延

SingleOutputStreamOperator> orderDSWithWatemark=formatData    .assignTimestampsAndWatermarks( // 设置watermark  watemark = 最大事件时间 - 最大延迟或乱序时间    WatermarkStrategy.>forBoundedOutOfOrderness(Duration.ofSeconds(3)) //指定maxOutOfOrderness最大无序度时间即最大延迟时间/乱序时间    .withTimestampAssigner((data,timestamp) -> Long.parseLong(DateUtil.dateToUTC(data.f0))*1000)  //时间为毫秒级);

• 设定窗口大小和处理逻辑

SingleOutputStreamOperator> result=orderDSWithWatemark.keyBy(one -> one.f1)      .window(TumblingEventTimeWindows.of(Time.seconds(10))) // 设定窗口大小//              .allowedLateness(Time.seconds(1))  //延时处理时间//              .sideOutputLateData(lateOutputTag)   //侧输出.reduce(new ReduceFunction>() { // 处理逻辑    private static final long serialVersionUID = -6695049408336015245L;    @Override    public Tuple3 reduce(Tuple3 value1,        Tuple3 value2) throws Exception {      Tuple3 data = new Tuple3();      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版本时间、水位线的介绍和用法"有了更深的了解，不妨来实际操作一番吧！这里是网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！