Flink如何处理乱序数据

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

1　Time

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

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

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

$F:\paiban\电子书\19-10-51678\54-55-web-resources\image\0801.jpg{84%}$

图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所示。

$F:\paiban\电子书\19-10-51678\54-55-web-resources\image\图像141521.png{85%}$

图2　代码设置使用哪种时间

2　Flink如何处理乱序数据

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

2.1　Watermark

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

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

$F:\paiban\电子书\19-10-51678\54-55-web-resources\image\0803.jpg{67%}$

图3　有序的Stream中的Watermark

无序的Stream中的Watermark，如图4所示。

$F:\paiban\电子书\19-10-51678\54-55-web-resources\image\0804.jpg{67%}$

图4　无序的Stream中的Watermark

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

$F:\paiban\电子书\19-10-51678\54-55-web-resources\image\0805.jpg{85%}$

图5　多并行度Stream中的watermark

注意：在多并行度的情况下，Watermark会有一个对齐机制，这个对齐机制会取所有Channel中最小的Watermark，图8.5中的14和29这两个Watermark的最终取值为14。

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所示。

$F:\paiban\电子书\19-10-51678\54-55-web-resources\image\图像141550.png{80%}$

图6　With Periodic Watermarks的使用

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

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

$F:\paiban\电子书\19-10-51678\54-55-web-resources\image\图像141560.png{75%}$

图7　Watermark的使用

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

3.1　实现Watermark的相关代码

1．程序说明

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

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）接收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，输入第一条数据。

[root@hadoop100 soft]# nc -l 9000
0001,1538359882000

输出的结果如图8所示。

{100%}

图8　Watermark输出的结果

为了查看方便，我们把输入内

容汇总到表格中，如表1所示。

表1　Watermark输出的结果

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

[root@hadoop100 soft]# nc -l 9000
0001,1538359882000
0001,1538359886000

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

$F:\paiban\电子书\19-10-51678\54-55-web-resources\image\图像141721.png{100%}$

图9　Watermark输出的结果

我们再次汇总，如表2所示。

继续输入。

[root@hadoop100 soft]# nc -l 9000
0001,1538359882000
0001,1538359886000
0001,1538359892000

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

$F:\paiban\电子书\19-10-51678\54-55-web-resources\image\图像141753.png{100%}$

图10　Watermark输出的结果

本文截选自《Flink入门与实战》

徐葳著

深入浅出展现Flink技术精髓
力求详细而完整地描述Flink大数据项目实战
从零开始快速掌握Flink的基本原理和核心功能，51CTO学院网课配套教材

本书旨在帮助读者从零开始快速掌握Flink的基本原理与核心功能。本书首先介绍了Flink的基本原理和安装部署，并对Flink中的一些核心API进行了详细分析。然后配套对应的案例分析，分别使用Java代码和Scala代码实现案例。最后通过两个项目演示了Flink在实际工作中的一些应用场景，帮助读者快速掌握Flink开发。
学习本书需要大家具备一些大数据的基础知识，比如Hadoop、Kafka、Redis、Elasticsearch等框架的基本安装和使用。本书也适合对大数据实时计算感兴趣的读者阅读。

学习本书需要大家具备一些大数据的基础知识，例如Hadoop、Kafka、Redis、Elasticsearch等框架的基本安装和使用。本书也适合对大数据实时计算感兴趣的爱好者阅读。

Flink如何处理乱序数据

1 Time

2 Flink如何处理乱序数据

2.1 Watermark

2.2 Watermark的生成方式

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

3.1 实现Watermark的相关代码