Flink的时间类型和WaterMark机制

1.1 需求背景

需求描述：每隔5秒，计算近10秒单词出现的次数。

1.1.1 TimeWindow实现

/**
 * 每隔5秒统计最近10秒的单词出现的次数
 */
public class TimeWindowWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataStream = env.socketTextStream("bigdata02", 1234);
        dataStream.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public void flatMap(String line,
                                Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] fields = line.split(",");
                for(String word:fields){
                    out.collect(Tuple2.of(word,1));
                }

            }
        }).keyBy(0).timeWindow(Time.seconds(10),Time.seconds(5))
                .sum(1)
                .print().setParallelism(1);

        env.execute("TimeWindowWordCount");

    }
}

1.1.2 ProcessWindowFunction

/**
 * 每隔5秒统计最近10秒的单词出现的次数
 */
public class TimeWindowWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataStream = env.socketTextStream("bigdata02", 1234);
        dataStream.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public void flatMap(String line,
                                Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] fields = line.split(",");
                for(String word:fields){
                    out.collect(Tuple2.of(word,1));
                }
                
            }
        }).keyBy(0).timeWindow(Time.seconds(10),Time.seconds(5))
                .process(new MySumProcessWindowFunction()) //foreach key,value -> sum -> key,value
                .print().setParallelism(1);

        env.execute("word count..");
    }

    /**
     * IN, OUT, KEY, W extends Window
     */
    public  static class MySumProcessWindowFunction extends
            ProcessWindowFunction<Tuple2<String,Integer>,Tuple2<String,Integer>
                    ,Tuple,TimeWindow>{
        FastDateFormat dataformat=FastDateFormat.getInstance("HH:mm:ss");
        @Override
        public void process(Tuple key, Context context,
                            Iterable<Tuple2<String, Integer>> elements,
                            Collector<Tuple2<String, Integer>> out) throws Exception {

            System.out.println("当前系统时间："+dataformat.format(System.currentTimeMillis()));
            System.out.println("窗口处理时间："+dataformat.format(context.currentProcessingTime()));
            System.out.println("窗口开始时间："+dataformat.format(context.window().getStart()));


            int sum=0;
            for (Tuple2<String,Integer> ele:elements){
                sum += 1;
            }
            out.collect(Tuple2.of(key.getField(0),sum));

            System.out.println("窗口结束时间："+dataformat.format(context.window().getEnd()));

            System.out.println("=====================================================");
        }
    }

根据每隔5秒执行近10秒的数据，Flink划分的窗口

[00:00:00, 00:00:05) [00:00:05, 00:00:10)
[00:00:10, 00:00:15) [00:00:15, 00:00:20)
[00:00:20, 00:00:25) [00:00:25, 00:00:30) 
[00:00:30, 00:00:35) [00:00:35, 00:00:40) 
[00:00:40, 00:00:45) [00:00:45, 00:00:50) 
[00:00:50, 00:00:55) [00:00:55, 00:01:00) 
[00:01:00, 00:01:05)  ...

1.1.3 Time的种类

针对stream数据中的时间，可以分为以下三种：
Event Time：事件产生的时间，它通常由事件中的时间戳描述。
Ingestion time：事件（日志，数据，消息）进入Flink的时间(不考虑)
Processing Time：事件被处理时当前系统的时间
默认情况下，我们使用的是Processing Time

案例演示：原始日志如下

2020-04-11 10:00:01,134 INFO executor.Executor: Finished task in state 0.0

这条数据进入Flink的时间是2020-04-11 20:00:00,102

到达window处理的时间为2020-04-11 20:00:01,100

2020-04-11 10:00:01,134 是Event time
2020-04-11 20:00:00,102 是Ingestion time
2020-04-11 20:00:01,100 是Processing time
思考：

如果我们想要统计每分钟内接口调用失败的错误日志个数，使用哪个时间才有意义？

1.2 Process Time Window（有序）

需求：每隔5秒计算近10秒的单词出现的次数
自定义source，模拟：第 13 秒的时候连续发送 2 个事件，第 16 秒的时候再发送 1 个事件

/**
 * 需求：每隔5秒计算最近10秒的单词次数
 */
public class WindowWordCountBySource {
    public static void main(String[] args) throws  Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataStream = env.addSource(new TestSource());
        dataStream.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public void flatMap(String line,
                                Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] fields = line.split(",");
                for (String word:fields){
                    out.collect(Tuple2.of(word,1));
                }
            }
        })
                .keyBy(0)
                .timeWindow(Time.seconds(10),Time.seconds(5))
                .process(new SumProcessFunction()).print().setParallelism(1);


        env.execute("WindowWordCountAndTime");

    }

    public static class TestSource implements
            SourceFunction<String>{
        FastDateFormat dateformat =  FastDateFormat.getInstance("HH:mm:ss");
        @Override
        public void run(SourceContext<String> cxt) throws Exception {

            String currTime = String.valueOf(System.currentTimeMillis());

            System.out.println(currTime);

            //这个操作是我为了保证是 10s的倍数。
            //我们执行的时间是10的倍数
            //10 20 30 40 50
            while(Integer.valueOf(currTime.substring(currTime.length() - 4)) > 100){
                currTime=String.valueOf(System.currentTimeMillis());
                continue;
            }

            System.out.println("开始发送事件的时间："+dateformat.format(System.currentTimeMillis()));
            TimeUnit.SECONDS.sleep(3);
            cxt.collect("hadoop");
            cxt.collect("hadoop");

            TimeUnit.SECONDS.sleep(3);
            cxt.collect("hadoop");

            TimeUnit.SECONDS.sleep(300000);

        }

        @Override
        public void cancel() {

        }
    }

    /**
     * IN
     * OUT
     * KEY
     * W extends Window
     *
     */
    public static class  SumProcessFunction
            extends ProcessWindowFunction<Tuple2<String,Integer>,Tuple2<String,Integer>,Tuple,TimeWindow>{

        FastDateFormat dataformat=FastDateFormat.getInstance("HH:mm:ss");
        @Override
        public void process(Tuple tuple, Context context,
                            Iterable<Tuple2<String, Integer>> allElements,
                            Collector<Tuple2<String, Integer>> out) {

            int count=0;
           for (Tuple2<String,Integer> e:allElements){
               count++;
           }
            out.collect(Tuple2.of(tuple.getField(0),count));

        }
    }
}

1.3 Process Time Window（无序）

自定义source，模拟：第 13 秒的时候连续发送 2 个事件，但是有一个事件确实在第13秒的发送出去了，另外一个事件因为某种原因在19秒的时候才发送出去，第 16 秒的时候再发送 1 个事件

/**
 * 需求：每隔5秒计算最近10秒的单词次数
 * 乱序
 */
public class WindowWordCountBySource2 {
    public static void main(String[] args) throws  Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataStream = env.addSource(new TestSource());
        dataStream.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public void flatMap(String line,
                                Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] fields = line.split(",");
                for (String word:fields){
                    out.collect(Tuple2.of(word,1));
                }
            }
        })
                .keyBy(0)
                .timeWindow(Time.seconds(10),Time.seconds(5))
                .process(new SumProcessFunction()).print().setParallelism(1);


        env.execute("WindowWordCountAndTime");

    }

    public static class TestSource implements
            SourceFunction<String>{
        FastDateFormat dateformat =  FastDateFormat.getInstance("HH:mm:ss");
        @Override
        public void run(SourceContext<String> cxt) throws Exception {
            String currTime = String.valueOf(System.currentTimeMillis());
            while(Integer.valueOf(currTime.substring(currTime.length() - 4)) > 100){
                currTime=String.valueOf(System.currentTimeMillis());
                continue;
            }
            System.out.println("开始发送事件的时间："+dateformat.format(System.currentTimeMillis()));

            //13
            TimeUnit.SECONDS.sleep(3);
            //实际上我们的数据是在13秒的时候生成的，只是19的时候被发送出去。
            String event="hadoop";

            cxt.collect(event);
         // cxt.collect(event);

            //16
            TimeUnit.SECONDS.sleep(3);
            cxt.collect("hadoop");
            //19
            TimeUnit.SECONDS.sleep(3);
            cxt.collect(event);

            TimeUnit.SECONDS.sleep(300990);

        }

        @Override
        public void cancel() {

        }
    }

    /**
     * IN
     * OUT
     * KEY
     * W extends Window
     *
     */
    public static class  SumProcessFunction
            extends ProcessWindowFunction<Tuple2<String,Integer>,Tuple2<String,Integer>,Tuple,TimeWindow>{

        FastDateFormat dataformat=FastDateFormat.getInstance("HH:mm:ss");
        @Override
        public void process(Tuple tuple, Context context,
                            Iterable<Tuple2<String, Integer>> allElements,
                            Collector<Tuple2<String, Integer>> out) {

            int count=0;
           for (Tuple2<String,Integer> e:allElements){
               count++;
           }
            out.collect(Tuple2.of(tuple.getField(0),count));

        }
    }
}

1.4 使用Event Time处理无序

使用Event Time处理

/**
 * 需求：每隔5秒计算最近10秒的单词次数
 */
public class WindowWordCountByEventTime {
    public static void main(String[] args) throws  Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //步骤一：设置时间类型，默认的是Processtime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> dataStream = env.addSource(new TestSource());
        dataStream.map(new MapFunction<String, Tuple2<String,Long>>() {
            @Override
            public Tuple2<String, Long> map(String line) throws Exception {
                String[] fields = line.split(",");
                    //key/value
                return new Tuple2<>(fields[0],Long.valueOf(fields[1]));
            }
            //作用：指定时间字段
        }).assignTimestampsAndWatermarks(new EventTimeExtractor())
                .keyBy(0)
                .timeWindow(Time.seconds(10),Time.seconds(5))
                .process(new SumProcessFunction()).print().setParallelism(1);


        env.execute("WindowWordCountAndTime");

    }

    public static class TestSource implements
            SourceFunction<String>{
        FastDateFormat dateformat =  FastDateFormat.getInstance("HH:mm:ss");
        @Override
        public void run(SourceContext<String> cxt) throws Exception {
            String currTime = String.valueOf(System.currentTimeMillis());
            while(Integer.valueOf(currTime.substring(currTime.length() - 4)) > 100){
                currTime=String.valueOf(System.currentTimeMillis());
                continue;
            }
            System.out.println("开始发送事件的时间："+dateformat.format(System.currentTimeMillis()));
            TimeUnit.SECONDS.sleep(3);
            //13
            String event="hadoop,"+System.currentTimeMillis();//时间

            cxt.collect(event);
           // cxt.collect(event);

            TimeUnit.SECONDS.sleep(3);//16
            cxt.collect("hadoop,"+System.currentTimeMillis());

            TimeUnit.SECONDS.sleep(3);
            //19
            cxt.collect(event);

            TimeUnit.SECONDS.sleep(3000);
            
        }

        @Override
        public void cancel() {

        }
    }


    private static class EventTimeExtractor
            implements AssignerWithPeriodicWatermarks<Tuple2<String,Long>>{
        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            return new Watermark(System.currentTimeMillis());
        }

        //指定时间
        @Override
        public long extractTimestamp(Tuple2<String, Long> element, long l) {
            return element.f1;
        }
    }

    /**
     * IN
     * OUT
     * KEY
     * W extends Window
     *
     */
    public static class  SumProcessFunction
            extends ProcessWindowFunction<Tuple2<String,Long>,Tuple2<String,Integer>,Tuple,TimeWindow>{

        FastDateFormat dataformat=FastDateFormat.getInstance("HH:mm:ss");
        @Override
        public void process(Tuple tuple, Context context,
                            Iterable<Tuple2<String, Long>> allElements,
                            Collector<Tuple2<String, Integer>> out) {

            int count=0;
           for (Tuple2<String,Long> e:allElements){
               count++;
           }
            out.collect(Tuple2.of(tuple.getField(0),count));

        }
    }
}

现在我们第三个window的结果已经计算准确了，但是我们还是没有彻底的解决问题。接下来就需要我们使用WaterMark机制来解决了。

1.5 使用【WaterMark】机制解决无序

/**
 * 需求：每隔5秒计算最近10秒的单词次数
 *
 * 引入watermark解决问题
 */
public class WindowWordCountByWaterMark {
    public static void main(String[] args) throws  Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //步骤一：设置时间类型，默认的是Processtime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> dataStream = env.addSource(new TestSource());
        dataStream.map(new MapFunction<String, Tuple2<String,Long>>() {
            @Override
            public Tuple2<String, Long> map(String line) throws Exception {
                String[] fields = line.split(",");
                return new Tuple2<>(fields[0],Long.valueOf(fields[1]));
            }
        }).assignTimestampsAndWatermarks(new EventTimeExtractor())
                .keyBy(0)
                .timeWindow(Time.seconds(10),Time.seconds(5))
                .process(new SumProcessFunction()).print().setParallelism(1);


        env.execute("WindowWordCountAndTime");

    }

    public static class TestSource implements
            SourceFunction<String>{
        FastDateFormat dateformat =  FastDateFormat.getInstance("HH:mm:ss");
        @Override
        public void run(SourceContext<String> cxt) throws Exception {
            String currTime = String.valueOf(System.currentTimeMillis());
            while(Integer.valueOf(currTime.substring(currTime.length() - 4)) > 100){
                currTime=String.valueOf(System.currentTimeMillis());
                continue;
            }
            System.out.println("开始发送事件的时间："+dateformat.format(System.currentTimeMillis()));
            TimeUnit.SECONDS.sleep(3);
            String event="hadoop,"+System.currentTimeMillis();
            cxt.collect(event);

            TimeUnit.SECONDS.sleep(3);
            cxt.collect("hadoop,"+System.currentTimeMillis());

            TimeUnit.SECONDS.sleep(3);
            cxt.collect(event);
            TimeUnit.SECONDS.sleep(3000);
            
        }

        @Override
        public void cancel() {

        }
    }


    private static class EventTimeExtractor implements AssignerWithPeriodicWatermarks<Tuple2<String,Long>>{
        //设置 5s的延迟（乱序）
        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            //System.out.println("water mark:......"+System.currentTimeMillis());
            return new Watermark(System.currentTimeMillis() - 5000);
        }

        @Override
        public long extractTimestamp(Tuple2<String, Long> element, long l) {
            return element.f1;
        }
    }

    /**
     * IN
     * OUT
     * KEY
     * W extends Window
     *
     */
    public static class  SumProcessFunction
            extends ProcessWindowFunction<Tuple2<String,Long>,Tuple2<String,Integer>,Tuple,TimeWindow>{

        FastDateFormat dataformat=FastDateFormat.getInstance("HH:mm:ss");
        @Override
        public void process(Tuple tuple, Context context,
                            Iterable<Tuple2<String, Long>> allElements,
                            Collector<Tuple2<String, Integer>> out) {

            int count=0;
           for (Tuple2<String,Long> e:allElements){
               count++;
           }
            out.collect(Tuple2.of(tuple.getField(0),count));

        }
    }
}

1.6 WaterMark机制

1.6.2 WaterMark的定义

我们知道，流处理从事件产生，到流经source，再到operator，中间是有一个过程和时间的。虽然大部分情况下，流到operator的数据都是按照事件产生的时间顺序来的，但是也不排除由于网络延迟等原因，导致乱序的产生，特别是使用kafka的话，多个分区的数据无法保证有序。所以在进行window计算的时候，我们又不能无限期的等下去，必须要有个机制来保证一个特定的时间后，必须触发window去进行计算了。这个特别的机制，就是watermark，watermark是用于处理乱序事件的。watermark可以翻译为水位线

有序的流的watermarks

无序的流的watermarks

多并行度流的watermarks

1.6.3 小需求

得到并打印每隔 3 秒钟统计前 3 秒内的相同的 key 的所有的事件
允许最大乱序时间4s

/**
 * 需求：3秒一个窗口，把相同 key合并起来
 *
 * hadoop,1461756862000
 * hadoop,1461756866000
 * hadoop,1461756872000
 * hadoop,1461756873000
 * hadoop,1461756874000
 * hadoop,1461756876000
 * hadoop,1461756877000
 *
 *
 * window + watermark  观察窗口是如何被触发？
 *
 * 可以解决乱序问题
 *
 *
 */
public class WindowWordCountByWaterMark2 {
    public static void main(String[] args) throws  Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //步骤一：设置时间类型
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //checkpoint

        DataStreamSource<String> dataStream = env.socketTextStream("bigdata02", 1234);
        dataStream.map(new MapFunction<String, Tuple2<String,Long>>() {
            @Override
            public Tuple2<String, Long> map(String line) throws Exception {
                String[] fields = line.split(",");
                return new Tuple2<>(fields[0],Long.valueOf(fields[1]));
            }
            //步骤二：获取数据里面的event Time
        }).assignTimestampsAndWatermarks(new EventTimeExtractor() )
                .keyBy(0)
                .timeWindow(Time.seconds(3))
                .process(new SumProcessWindowFunction())
                .print().setParallelism(1);

        env.execute("WindowWordCountByWaterMark2");

    }

    /**
     * IN, OUT, KEY, W
     * IN：输入的数据类型
     * OUT：输出的数据类型
     * Key：key的数据类型（在Flink里面，String用Tuple表示）
     * W：Window的数据类型
     */
    public static class SumProcessWindowFunction extends
            ProcessWindowFunction<Tuple2<String,Long>,String,Tuple,TimeWindow> {
        FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");
        /**
         * 当一个window触发计算的时候会调用这个方法
         * @param tuple key
         * @param context operator的上下文
         * @param elements 指定window的所有元素
         * @param out 用户输出
         */
        @Override
        public void process(Tuple tuple, Context context,
                            Iterable<Tuple2<String, Long>> elements,
                            Collector<String> out) {
            System.out.println("处理时间：" + dateFormat.format(context.currentProcessingTime()));
            System.out.println("window start time : " + dateFormat.format(context.window().getStart()));

            List<String> list = new ArrayList<>();
            for (Tuple2<String, Long> ele : elements) {
                list.add(ele.toString() + "|" + dateFormat.format(ele.f1));
            }
            out.collect(list.toString());
            System.out.println("window end time  : " + dateFormat.format(context.window().getEnd()));

        }
    }

    private static class EventTimeExtractor
            implements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {
        FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");
        //3s 计算时间最大的一个值。
        private long currentMaxEventTime=0L;
        private long maxOufOfOrderness=10000;//最大乱序时间

        //30
        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            //watermark的计算的逻辑
            //如何计算出来一个窗口里面的watermark
            return new Watermark(currentMaxEventTime - maxOufOfOrderness);
        }

        @Override
        public long extractTimestamp(Tuple2<String, Long> element, long timeStamp) {
            Long currentElementTime = element.f1;


            currentMaxEventTime = Math.max(currentMaxEventTime,currentElementTime);
//
            System.out.println("event = " + element
                    + "|" + dateFormat.format(element.f1) // Event Time
                    + "|" + dateFormat.format(currentMaxEventTime)  // Max Event Time
                    + "|" + dateFormat.format(getCurrentWatermark().getTimestamp())); // Current Watermark

            return currentElementTime;
        }
    }
}

[00:00:00, 00:00:03) [00:00:03, 00:00:06)
[00:00:06, 00:00:12) [00:00:12, 00:00:15)
[00:00:15, 00:00:18) [00:00:18, 00:00:21) 
[00:00:21, 00:00:24) [00:00:24, 00:00:27) 
[00:00:27, 00:00:30) [00:00:30, 00:00:33) 
[00:00:33, 00:00:36) [00:00:36, 00:00:39) 
[00:00:42, 00:00:45)  ...

Key	Event Time	currentMaxTimestamep	currentWatermark	window start time	window end time
000001	`19:34:22`	19:34:22	19:34:12
000001	`19:34:26`	19:34:26	19:34:16
000001	19:34:32	19:34:32	19:34:22
000001	19:34:33	19:34:33	19:34:23
000001	19:34:34	19:34:34	`19:34:24`	[`19:34:21`	`19:34:24`)
000001	19:34:36	19:34:36	19:34:26
000001	19:34:37	19:34:37	`19:34:27`	[`19:34:24`	`19:34:27`)

总结：window触发的时间

1. watermark 时间 >= window_end_time
2. 在 [window_start_time, window_end_time) 区间中有数据存在，注意是左闭右开的区间，而且是以 event time 来计算的

1.6.5 WaterMark+Window 处理乱序事件

/**
 * 需求：3秒一个窗口，进行单词计数
 *
 * 迟到太多的数据：默认就丢弃了
 *
 * 000001,1461756879000
 * 000001,1461756871000
 * 000001,1461756883000
 */
public class WindowWordCountByWaterMark3 {
    public static void main(String[] args) throws  Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //步骤一：设置时间类型
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> dataStream = env.socketTextStream("bigdata02", 1234);
        dataStream.map(new MapFunction<String, Tuple2<String,Long>>() {
            @Override
            public Tuple2<String, Long> map(String line) throws Exception {
                String[] fields = line.split(",");
                return new Tuple2<>(fields[0],Long.valueOf(fields[1]));
            }
            //步骤二：获取数据里面的event Time
        }).assignTimestampsAndWatermarks(new EventTimeExtractor())
                .keyBy(0)
                .timeWindow(Time.seconds(3))
               // .allowedLateness(Time.seconds(2)) // 允许事件迟到 2 秒
                .process(new SumProcessWindowFunction())
                .print().setParallelism(1);

        env.execute("WindowWordCountByWaterMark3");

    }

    /**
     * IN, OUT, KEY, W
     * IN：输入的数据类型
     * OUT：输出的数据类型
     * Key：key的数据类型（在Flink里面，String用Tuple表示）
     * W：Window的数据类型
     */
    public static class SumProcessWindowFunction extends
            ProcessWindowFunction<Tuple2<String,Long>,String,Tuple,TimeWindow> {
        FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");
        /**
         * 当一个window触发计算的时候会调用这个方法
         * @param tuple key
         * @param context operator的上下文
         * @param elements 指定window的所有元素
         * @param out 用户输出
         */
        @Override
        public void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements,
                            Collector<String> out) {
            System.out.println("处理时间：" + dateFormat.format(context.currentProcessingTime()));
            System.out.println("window start time : " + dateFormat.format(context.window().getStart()));

            List<String> list = new ArrayList<>();
            for (Tuple2<String, Long> ele : elements) {
                list.add(ele.toString() + "|" + dateFormat.format(ele.f1));
            }
            out.collect(list.toString());
            System.out.println("window end time  : " + dateFormat.format(context.window().getEnd()));

        }
    }

    private static class EventTimeExtractor
            implements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {
        FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");
        private long currentMaxEventTime=0L;
        private long maxOufOfOrderness=10000;//最大乱序时间
        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            return new Watermark(currentMaxEventTime - maxOufOfOrderness);
        }

        @Override
        public long extractTimestamp(Tuple2<String, Long> element, long timeStamp) {
            Long currentElementTime = element.f1;
            currentMaxEventTime = Math.max(currentMaxEventTime,currentElementTime);

            System.out.println("event = " + element
                    + "|" + dateFormat.format(element.f1) // Event Time
                    + "|" + dateFormat.format(currentMaxEventTime)  // Max Event Time
                    + "|" + dateFormat.format(getCurrentWatermark().getTimestamp())); // Current Watermark

            return currentElementTime;
        }
    }
}

Key	Event Time	currentMaxTimestamep	currentWatermark	window start time	window end time
000001	19:34:39	19:34:39	19:34:29
000001	`19:34:31`	19:34:39	19:34:29
000001	19:34:43	19:34:43	19:34:33	[`19:34:30`	`19:34:33`)

1.6.6 迟到太多的事件

1. 丢弃，这个是默认的处理方式
2. allowedLateness 指定允许数据延迟的时间(我们一般不选择)
3. sideOutputLateData 收集迟到的数据(修正实时的数据)

丢弃

重启程序，做测试。

输入数据：

000001,1461756870000 
000001,1461756883000 
000001,1461756870000 
000001,1461756871000 
000001,1461756872000

Key	Event Time	currentMaxTimestamep	currentWatermark	window start time	window end time
000001	19:34:30	19:34:30	19:34:20
000001	19:34:43	19:34:43	19:34:33	[19:34:30	19:34:33)
000001	19:34:30	19:34:43	19:34:33	[19:34:30	19:34:33)
000001	19:34:31	19:34:43	19:34:33	[19:34:30	19:34:33)
000001	19:34:32	19:34:43	19:34:33	[19:34:30	19:34:33)

发现迟到太多数据就会被丢弃

指定允许再次迟到的时间

).assignTimestampsAndWatermarks(new EventTimeExtractor() )
               .keyBy(0)
                .timeWindow(Time.seconds(3)) 
               .allowedLateness(Time.seconds(2)) // 允许事件迟到 2 秒
                .process(new SumProcessWindowFunction())
                .print().setParallelism(1);

输入数据

000001,1461756870000 
000001,1461756883000

000001,1461756870000 
000001,1461756871000 
000001,1461756872000
000001,1461756884000

000001,1461756870000 
000001,1461756871000 
000001,1461756872000
000001,1461756885000

000001,1461756870000 
000001,1461756870000 
000001,1461756872000

1. 当我们设置允许迟到 2 秒的事件，第一次 window 触发的条件是 watermark >= window_end_time
2. 第二次(或者多次)触发的条件是 watermark < window_end_time + allowedLateness

收集迟到的数据

/**
 * 需求：3秒一个窗口，进行单词计数
 *
 * 指定延迟时间
 * 收集延迟数据
 */
public class WindowWordCountByWaterMark4 {
    public static void main(String[] args) throws  Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //步骤一：设置时间类型
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //设置waterMark产生的周期为1s
        env.getConfig().setAutoWatermarkInterval(1000);

        DataStreamSource<String> dataStream = env.socketTextStream("bigdata02", 1234);

        //步骤一：声明这样的一个数据结构
        OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("late-date"){};

        SingleOutputStreamOperator<String> result = dataStream.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String line) throws Exception {
                String[] fields = line.split(",");
                return new Tuple2<>(fields[0], Long.valueOf(fields[1]));
            }
            //步骤二：获取数据里面的event Time
        }).assignTimestampsAndWatermarks(new EventTimeExtractor())
                .keyBy(0)
                .timeWindow(Time.seconds(3))
             //   .allowedLateness(Time.seconds(2)) // 允许事件迟到 2 秒
                .sideOutputLateData(outputTag) //保留迟到太多的数据
                .process(new SumProcessWindowFunction());
        //打印结果数据
        result.print();


        /**
         *
         * 处理延迟的数据：
         * 1. 准备另外一个延迟的topic（Kafka，磁盘，数据库）
         *
         * 不会有太多，如果没有太多，其实结果影响不大。
         *
         *
         * 实时的任务，大多数的情况就是为了看这个趋势而已。
         *
         *
         */
        result.getSideOutput(outputTag).map(new MapFunction<Tuple2<String,Long>, String>() {
            @Override
            public String map(Tuple2<String, Long> stringLongTuple2) throws Exception {
                return "迟到数据："+stringLongTuple2.toString();
            }
        }).print();

        env.execute("WindowWordCountByWaterMark3");

    }

    /**
     * IN, OUT, KEY, W
     * IN：输入的数据类型
     * OUT：输出的数据类型
     * Key：key的数据类型（在Flink里面，String用Tuple表示）
     * W：Window的数据类型
     */
    public static class SumProcessWindowFunction extends
            ProcessWindowFunction<Tuple2<String,Long>,String,Tuple,TimeWindow> {
        FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");
        /**
         * 当一个window触发计算的时候会调用这个方法
         * @param tuple key
         * @param context operator的上下文
         * @param elements 指定window的所有元素
         * @param out 用户输出
         */
        @Override
        public void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements,
                            Collector<String> out) {
            System.out.println("处理时间：" + dateFormat.format(context.currentProcessingTime()));
            System.out.println("window start time : " + dateFormat.format(context.window().getStart()));

            List<String> list = new ArrayList<>();
            for (Tuple2<String, Long> ele : elements) {
                list.add(ele.toString() + "|" + dateFormat.format(ele.f1));
            }
            out.collect(list.toString());
            System.out.println("window end time  : " + dateFormat.format(context.window().getEnd()));

        }
    }

    private static class EventTimeExtractor
            implements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {
        FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");
        private long currentMaxEventTime=0L;
        private long maxOufOfOrderness=10000;//最大乱序时间
        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            return new Watermark(currentMaxEventTime - maxOufOfOrderness);
        }

        @Override
        public long extractTimestamp(Tuple2<String, Long> element, long timeStamp) {
            Long currentElementTime = element.f1;
            currentMaxEventTime = Math.max(currentMaxEventTime,currentElementTime);

            System.out.println("event = " + element
                    + "|" + dateFormat.format(element.f1) // Event Time
                    + "|" + dateFormat.format(currentMaxEventTime)  // Max Event Time
                    + "|" + dateFormat.format(getCurrentWatermark().getTimestamp())); // Current Watermark

            return currentElementTime;
        }
    }
}

1.7 多并行度下的WaterMark

一个window可能会接受到多个waterMark，我们以小的为准。

/**
 * 需求：3秒一个窗口，进行单词计数
 *
 * 多并行度下多watermark
 *
 * 000001,1461756870000
 * 000001,1461756883000
 * 000001,1461756888000
 *
 */
public class WindowWordCountByWaterMark5 {
    public static void main(String[] args) throws  Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //步骤一：设置时间类型
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        env.setParallelism(2);

        DataStreamSource<String> dataStream = env.socketTextStream("bigdata02", 1234);


        OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("late-date"){};

        SingleOutputStreamOperator<String> result = dataStream.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String line) throws Exception {
                String[] fields = line.split(",");
                return new Tuple2<>(fields[0], Long.valueOf(fields[1]));
            }
            //步骤二：获取数据里面的event Time
        }).assignTimestampsAndWatermarks(new EventTimeExtractor())
                .keyBy(0)
                .timeWindow(Time.seconds(3))
//                .allowedLateness(Time.seconds(2)) // 允许事件迟到 2 秒
                .sideOutputLateData(outputTag) //保留迟到太多的数据
                .process(new SumProcessWindowFunction());
        //打印结果数据
        result.print();



        result.getSideOutput(outputTag).map(new MapFunction<Tuple2<String,Long>, String>() {
            @Override
            public String map(Tuple2<String, Long> stringLongTuple2) throws Exception {
                return "迟到数据："+stringLongTuple2.toString();
            }
        }).print();

        env.execute("WindowWordCountByWaterMark3");

    }

    /**
     * IN, OUT, KEY, W
     * IN：输入的数据类型
     * OUT：输出的数据类型
     * Key：key的数据类型（在Flink里面，String用Tuple表示）
     * W：Window的数据类型
     */
    public static class SumProcessWindowFunction extends
            ProcessWindowFunction<Tuple2<String,Long>,String,Tuple,TimeWindow> {
        FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");
        /**
         * 当一个window触发计算的时候会调用这个方法
         * @param tuple key
         * @param context operator的上下文
         * @param elements 指定window的所有元素
         * @param out 用户输出
         */
        @Override
        public void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements,
                            Collector<String> out) {
            System.out.println("处理时间：" + dateFormat.format(context.currentProcessingTime()));
            System.out.println("window start time : " + dateFormat.format(context.window().getStart()));

            List<String> list = new ArrayList<>();
            for (Tuple2<String, Long> ele : elements) {
                list.add(ele.toString() + "|" + dateFormat.format(ele.f1));
            }
            out.collect(list.toString());
            System.out.println("window end time  : " + dateFormat.format(context.window().getEnd()));

        }
    }

    private static class EventTimeExtractor
            implements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {
        FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");
        private long currentMaxEventTime=0L;
        private long maxOufOfOrderness=10000;//最大乱序时间
        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            return new Watermark(currentMaxEventTime - maxOufOfOrderness);
        }

        @Override
        public long extractTimestamp(Tuple2<String, Long> element, long timeStamp) {
            Long currentElementTime = element.f1;
            currentMaxEventTime = Math.max(currentMaxEventTime,currentElementTime);

            long id = Thread.currentThread().getId();

            System.out.println("当前线程id="+id+" event = " + element
                    + "|" + dateFormat.format(element.f1) // Event Time
                    + "|" + dateFormat.format(currentMaxEventTime)  // Max Event Time
                    + "|" + dateFormat.format(getCurrentWatermark().getTimestamp())); // Current Watermark

            return currentElementTime;
        }
    }
}

ID为58的线程有两个WaterMark：20,38
那么38这个会替代20，所以ID为58的线程的WaterMark是38
然后ID为57的线程的WaterMark是33，而ID为58是WaterMark是38，会在里面求一个小的值作为 waterMark，就是33，这个时候会触发Window为30-33的窗口，那这个窗口里面就有 (000001,1461756870000)这条数据。

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 206,214评论 6赞 481
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 88,307评论 2赞 382
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 152,543评论 0赞 341
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 55,221评论 1赞 279
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 64,224评论 5赞 371
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 49,007评论 1赞 284
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,313评论 3赞 399
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,956评论 0赞 259
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 43,441评论 1赞 300
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,925评论 2赞 323
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,018评论 1赞 333
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,685评论 4赞 322
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,234评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,240评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,464评论 1赞 261
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,467评论 2赞 352
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,762评论 2赞 345