窗口入门一（滚动窗口TumblingEventTimeWindows）

窗口的作用是为了周期性的获取数据，将原始的数据流切分为多个窗口，但是因为数据传输受网络影响造成数据的乱序和迟到数据，这时候就要提及两个概念

EventTime 数据生产时间，由数据携带（生产使用）
ProcessingTime 数据处理时间（生产极少使用）
IngestionTime 数据进入程序时间（生产极少使用）
Watermark 水位线，flink的一种特殊标记，记录某个时间点，由flink产生，单调递增，所谓的水位线更多的是为了预测迟到数据

简单示例代码

滚动窗口TumblingEventTimeWindows 滚动的也就是不会出现重复数据

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
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 org.flink.demo.common.source.DataSource;
import org.flink.demo.module.WaterMarkData;

import java.time.Duration;
import java.util.Iterator;

public class Flink03_Transform_Window {
    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //数据生产在 Flink入门系列2 keyby reduce process使用 中
        DataSource dataSource = new DataSource();

        DataStreamSource<WaterMarkData> markDataDataStreamSource = env.addSource(dataSource);

        //这里举例两个 forMonotonousTimestamps 和  forBoundedOutOfOrderness
        // 指定时间戳
//        SingleOutputStreamOperator<WaterMarkData> timestampsAndWatermarks = markDataDataStreamSource
//                .assignTimestampsAndWatermarks(WatermarkStrategy
//                        .<WaterMarkData>forMonotonousTimestamps() //针对时间戳单调上升的
//                        .withTimestampAssigner((t, s) -> {
//                            return t.getTs();
//                        }));
        SingleOutputStreamOperator<WaterMarkData> timestampsAndWatermarks = markDataDataStreamSource
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WaterMarkData>forBoundedOutOfOrderness(Duration.ofSeconds(0))//针对乱序时间 ,参数指定延迟时间
                        .withTimestampAssigner((t, s) -> {
                            return t.getTs(); //指定生产时间
                        }));
        SingleOutputStreamOperator<Integer> streamWindowOperator = timestampsAndWatermarks
                .keyBy(WaterMarkData::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))  //指定滚动窗口 和窗口大小
                .process(
                        /**
                         * 四个参数意义分别是，输入数据类型 输出数据类型 key 值类型 window 窗口类型
                         */
                        new ProcessWindowFunction<WaterMarkData, Integer, Integer, TimeWindow>() {
                            @Override
                            public void process(Integer integer, ProcessWindowFunction<WaterMarkData, Integer, Integer, TimeWindow>.Context context, Iterable<WaterMarkData> elements, Collector<Integer> out) throws Exception {
                                //这里就简单处理算年龄之和了
                                Iterator<WaterMarkData> elementIte = elements.iterator();
                                int ageSum = 0;
                                while (elementIte.hasNext()) {
                                    int age = elementIte.next().getAge();
                                    System.out.println("当前年龄：" + age);
                                    ageSum += age;
                                }
                                out.collect(ageSum);
                            }
                        });
        streamWindowOperator.print();
        env.execute();
    }
}