Flink如何处理乱序数据？

小灰灰 2022-10-31 10:47 271阅读 0赞

参考文章：[Flink如何处理乱序数据？][Flink]

本章主要针对Flink Time中的Event Time、Ingestion Time、Processing Time以及Watermark进行详细讲解。

# 1 Time #

Stream数据中的Time（时间）分为以下3种。

>  *  Event Time：事件产生的时间，它通常由事件中的时间戳描述。
>  *  Ingestion Time：事件进入Flink的时间。
>  *  Processing Time：事件被处理时当前系统的时间。

这几种时间的对应关系如图1所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70][] 图1　Flink中的3种Time之间的关系

**假设原始日志如下。**

> 2019-01-10 10:00:01,134 INFO executor.Executor: Finished task in state 0.0

> 这条数据进入Flink的时间是2019-01-10 20:00:00,102。
> 
> 到达Window处理的时间为2019-01-10 20:00:01,100。
> 
> 如果我们想要统计每分钟内接口调用失败的错误日志个数，使用哪个时间才有意义？因为数据有可能出现延迟，所以使用数据进入Flink的时间或者Window处理的时间，其实是没有意义的，此时使用原始日志中的时间才是有意义的，那才是数据产生的时间。

我们在Flink的Stream程序中处理数据时，默认使用的是哪个时间呢？如何修改呢？默认情况下，Flink在Stream程序中处理数据使用的时间是ProcessingTime，想要修改使用时间可以使用setStreamTimeCharacteristic()，代码如图2所示。

![20191028084410681.png][]

# 2 Flink如何处理乱序数据 #

> 在使用EventTime处理Stream数据的时候会遇到数据乱序的问题，流处理从Event（事件）产生，流经Source，再到Operator，这中间需要一定的时间。虽然大部分情况下，传输到Operator的数据都是按照事件产生的时间顺序来的，但是也不排除由于网络延迟等原因而导致乱序的产生，特别是使用Kafka的时候，多个分区之间的数据无法保证有序。因此，在进行Window计算的时候，不能无限期地等下去，必须要有个机制来保证在特定的时间后，必须触发Window进行计算，这个特别的机制就是Watermark。Watermark是用于处理乱序事件的。

### 2.1 Watermark ###

Watermark可以翻译为水位线，有3种应用场景。

*  有序的Stream中的Watermark，如图3所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70 1][]

*  多并行度Stream中的Watermark，如图5所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70 2][]

## 2.2　Watermark的生成方式 ##

通常情况下，在接收到Source的数据后，应该立刻生成Watermark，但是也可以在应用简单的Map或者Filter操作后再生成Watermark。

> **注意：**如果指定多次Watermark，后面指定的值会覆盖前面的值。

Watermark的生成方式有两种。

**1．With Periodic Watermarks**

>  *  周期性地触发Watermark的生成和发送，默认是100ms。
>  *  每隔*N*秒自动向流里注入一个Watermark，时间间隔由ExecutionConfig.setAutoWatermarkInterval决定。每次调用getCurrentWatermark方法，如果得到的Watermark不为空并且比之前的大，就注入流中。
>  *  可以定义一个最大允许乱序的时间，这种比较常用。
>  *  实现AssignerWithPeriodicWatermarks接口。

**2．With Punctuated Watermarks**

>  *  基于某些事件触发Watermark的生成和发送。
>  *  基于事件向流里注入一个Watermark，每一个元素都有机会判断是否生成一个Watermark。如果得到的Watermark不为空并且比之前的大，就注入流中。
>  *  实现AssignerWithPunctuatedWatermarks接口。

第1种方式比较常用，所以在这里我们使用第1种方式进行分析。

参考官网文档中With Periodic Watermarks的使用方法，如图6所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70 3][] 图6　With Periodic Watermarks的使用

图6所示代码中的extractTimestamp方法是从数据本身中提取EventTime。getCurrentWatermar方法是获取当前水位线，利用currentMaxTimestamp - maxOutOfOrderness。maxOutOfOrderness表示是允许数据的最大乱序时间。

在这里也需要实现接口AssignerWithPeriodicWatermarks，参考代码如图7所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70 4][]

# 3　EventTime+Watermark解决乱序数据的案例详解 #

## 3.1　实现Watermark的相关代码 ##

### **3.1.1．程序说明** ###

> 首先通过Socket模拟接收数据，然后使用map函数进行处理，接着调用assignTimestampsAnd-Watermarks方法抽取timestamp并生成Watermark，最后调用Window打印信息来验证Window被触发的时机。

### 3.1.2．代码实现 ###

package xuwei.tech.streaming.streamApiDemo;
    
    import org.apache.Flink.api.common.functions.MapFunction;
    import org.apache.Flink.api.java.tuple.Tuple;
    import org.apache.Flink.api.java.tuple.Tuple2;
    import org.apache.Flink.streaming.api.TimeCharacteristic;
    import org.apache.Flink.streaming.api.DataStream.DataStream;
    import org.apache.Flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.Flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
    import org.apache.Flink.streaming.api.functions.windowing.WindowFunction;
    import org.apache.Flink.streaming.api.watermark.Watermark;
    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.annotation.Nullable;
    import java.text.SimpleDateFormat;
    import java.util.ArrayList;
    import java.util.Collections;
    import java.util.Iterator;
    import java.util.List;
    
    /**
     * Watermark案例
     * Created by xuwei.tech
     */
    public class StreamingWindowWatermark {
    
        public static void main(String[] args) throws Exception {
            //定义Socket的端口号
            int port = 9000;
           
            //获取运行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            
            //设置使用EventTime，默认使用processtime
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            
            //设置并行度为1,默认并行度是当前机器的CPU数量
            env.setParallelism(1);
    
            //连接Socket获取输入的数据
            DataStream<String> text = env.socketTextStream("hadoop100", port, "\n");
    
            //解析输入的数据
            DataStream<Tuple2<String, Long>> inputMap = text.map(new MapFunction<String, Tuple2<String, Long>>() {
                @Override
                public Tuple2<String, Long> map(String value) throws Exception {
                    String[] arr = value.split(",");
                    return new Tuple2<>(arr[0], Long.parseLong(arr[1]));
                }
            });
    
            //抽取timestamp和生成Watermark
            DataStream<Tuple2<String, Long>> waterMarkStream = inputMap.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {
    
                Long currentMaxTimestamp = 0L;
                final Long maxOutOfOrderness = 10000L;// 最大允许的乱序时间是10s
    
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    
                /**
                 * 定义生成Watermark的逻辑
                 * 默认100ms被调用一次
                 */
                @Nullable
                @Override
                public Watermark getCurrentWatermark() {
                    return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
                }
    
                //定义如何提取timestamp
                @Override
                public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
                    long timestamp = element.f1;
                    currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
                    System.out.println("key:" + element.f0 + ",eventtime:[" + element.f1 + "|" + sdf.format(element.f1) + "],currentMaxTimestamp:[" + currentMaxTimestamp + "|" +
                            sdf.format(currentMaxTimestamp) + "],watermark:[" + getCurrentWatermark().getTimestamp() + "|" + sdf.format(getCurrentWatermark().getTimestamp()) + "]");
                    return timestamp;
                }
            });
    
            //分组，聚合
            DataStream<String> window = waterMarkStream.keyBy(0)
                    .window(TumblingEventTimeWindows.of(Time.seconds(3)))//按照消息的EventTime分配窗口，和调用Time Window效果一样
                    .apply(new WindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
                        /**
                         * 对Window内的数据进行排序，保证数据的顺序
                         * @param tuple
                         * @param window
                         * @param input
                         * @param out
                         * @throws Exception
                         */
                        @Override
                        public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2
                                <String, Long>> input, Collector<String> out) throws Exception {
                            String key = tuple.toString();
                            List<Long> arrarList = new ArrayList<Long>();
                            Iterator<Tuple2<String, Long>> it = input.iterator();
                            while (it.hasNext()) {
                                Tuple2<String, Long> next = it.next();
                                arrarList.add(next.f1);
                            }
                            Collections.sort(arrarList);
                            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                            String result = key + "," + arrarList.size() + "," + sdf.format(arrarList.get(0)) + "," + sdf.format(arrarList.get(arrarList.size() - 1))
                                    + "," + sdf.format(window.getStart()) + "," + sdf.format(window.getEnd());
                            out.collect(result);
                        }
                    });
            //测试，把结果打印到控制台即可
            window.print();
    
            //注意：因为Flink是懒加载的，所以必须调用execute方法，这样上面的代码才会执行
            env.execute("eventtime-watermark");
        }

### 3.1.3．程序详解 ###

> （1）接收Socket数据。
> 
> （2）将每行数据按照逗号分隔，每行数据调用Map转换成Tuple<String,Long>类型。其中Tuple中的第1个元素代表具体的数据，第2个元素代表数据的EventTime。
> 
> （3）抽取Timestamp，生成Watermark，允许的最大乱序时间是10s，并打印（Key，EventTime，CurrentMaxTimestamp，Watermark）等信息。
> 
> （4）分组聚合，Window窗口大小为3s，输出（Key，窗口内元素个数，窗口内最初元素进入的时间，窗口内最后元素进入的时间，窗口自身开始时间，窗口自身结束时间）。

## 3.2　通过数据跟踪Watermark的时间 ##

在这里重点查看Watermark和Timestamp的时间，通过数据的输出来确定Window的触发时机。

首先开启Socket，输入第一条数据。

> 1.  `[root@hadoop100 soft]# nc -l 9000`
> 2.  `0001,1538359882000`

输出的结果如图8(略)所示。

为了查看方便，我们把输入内容汇总到表格中，如表1所示。

表1　Watermark输出的结果

![20191028084729655.png][]

此时，Watermark的时间已经落后于CurrentMaxTimeStamp10s了，我们继续输入。

> 1.  `[root@hadoop100 soft]# nc -l 9000`
> 2.  `0001,1538359882000`
> 3.  `0001,1538359886000`

此时，输出的结果如图9（略）所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70 5][]

> 1.  `[root@hadoop100 soft]# nc -l 9000`
> 2.  `0001,1538359882000`
> 3.  `0001,1538359886000`
> 4.  `0001,1538359892000`

输出的结果内容如图10（略）所示。

[Flink]: https://blog.csdn.net/epubit17/article/details/102775009
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70]: /images/20221024/f64ffcdbcd7249ba8159174d7a0674bb.png
[20191028084410681.png]: /images/20221024/dce683a052f1424f9106eb65d5e978f2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70 1]: /images/20221024/447b04a6372d45ecb24ce1b9929889d4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70 2]: /images/20221024/04198192c224475eab77b5cc02599521.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70 3]: /images/20221024/5fc48693e60a43659a824796c1e62fee.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70 4]: /images/20221024/66ab32abfd054262a096a2e771ea225a.png
[20191028084729655.png]: /images/20221024/bc2543ed0d744057aeed6039f155b60e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2VwdWJpdDE3_size_16_color_FFFFFF_t_70 5]: /images/20221024/5dc838e5fa7e4e6a8f42fc1df2e02fca.png

发表评论取消回复

表情：

评论列表（有 0 条评论，271人围观）

还没有评论，来说两句吧...

相关阅读

相关 flink 如何分析及处理反压

在 Apache Flink 中，反压（Backpressure）是指当数据源产生的数据速度超过处理程序的处理能力时，处理程序需要向数据源发送信号来减慢数据产生的速度，以避免数

短命女/ 2024年03月16日 22:45/ 0 赞/ 50 阅读

相关如何分析及处理 Flink 反压？

如何分析及处理 Flink 反压？反压（backpressure）是实时计算应用开发中，特别是流式计算中，十分常见的问题。反压意味着数据管道中某个节点成为瓶颈，处理速率跟

一时失言乱红尘/ 2023年07月09日 02:26/ 0 赞/ 3 阅读

相关 Apache Flink 如何正确处理实时计算场景中的乱序数据

一、流式计算的未来在谷歌发表了 GFS、BigTable、Google MapReduce 三篇论文后，大数据技术真正有了第一次飞跃，Hadoop 生态系统逐渐发展起来

淡淡的烟草味﹌/ 2022年11月10日 03:43/ 0 赞/ 363 阅读

相关 Flink中处理乱序数据的三种方式

Flink中处理乱序数据的三种方式 1. 加水印 Flink中的时间语意WaterMark，以事件时间减去所允许的最大乱序时间作为水印，原理相当于多给了数据

系统管理员/ 2022年11月05日 14:58/ 0 赞/ 324 阅读

相关 Flink如何处理乱序数据？

参考文章：[Flink如何处理乱序数据？][Flink] 本章主要针对Flink Time中的Event Time、Ingestion Time、Processing Tim

小灰灰/ 2022年10月31日 10:47/ 0 赞/ 272 阅读

相关 Flink是如何处理反压的？

Flink 内部是基于 producer-consumer 模型来进行消息传递的，Flink的反压设计也是基于这个模型。Flink 使用了高效有界的分布式阻塞队列，就像 Jav

深碍√TFBOYSˉ_/ 2022年10月19日 12:59/ 0 赞/ 146 阅读

相关 IOS之取乱序数据最大值、最小值方法

第一种方法也是常规方法，就是设定一个默认值作为最大值，循环取比这个最大值还大的值并赋值给默认最大值，这样循环完成后这个默认最大值变量里面的值就是最大值了： float

我会带着你远行/ 2022年05月13日 03:08/ 0 赞/ 361 阅读

相关 Flink 水位线机制WaterMark实践处理乱序消息

1 水位线（WaterMark）是一个时间戳，等于当前到达的消息最大时间戳减去配置的延迟时间，水位线是单调递增的，如果有晚到达的早消息也不会更新水位线，因为消息最大时间戳没变

ゞ浴缸里的玫瑰/ 2021年09月24日 11:18/ 0 赞/ 294 阅读

相关 nodeJS与MySQL实现分页数据以及倒序数据

大家在做项目时肯定会遇到列表类的数据，如果在前台一下子展示，速度肯定很慢，那么我们可以分页展示，比如说100条数据，每10条一页，在需要的时候加载一页，这样速度肯定会变快了。

男娘i/ 2021年08月19日 19:30/ 0 赞/ 166 阅读

相关 js如何对乱序数组进行排序

js如何对乱序数组进行排序具体代码为：

小灰灰/ 2020年06月03日 12:48/ 0 赞/ 1113 阅读