Flink：基于时间驱动的滑动窗口使用

待我称王封你为后i 2024-03-23 11:54 65阅读 0赞

**时间滑动窗口：**

**滑动窗口是固定窗口的更广义的一种形式，滑动窗口由固定的窗口长度和滑动间隔组成**

**特点：窗口长度固定，可以有重叠**

**![b4ed7479b5714a439a798588d0f8e00f.png][]**

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.7.2</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>1.7.2</version>
        <scope>provided</scope>
    </dependency>

**发送**

> netcat下载：https://blog.csdn.net/qq\_38762390/article/details/115789281

nc -lk 7777

import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.tuple.Tuple;
    import org.apache.flink.api.java.tuple.Tuple1;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.KeyedStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.datastream.WindowedStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
    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 java.text.SimpleDateFormat;
     
    public class WindowDemo {
        public static void main(String[] args) throws Exception {
            // 获取数据源
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            String hostname = "";
            int port = 7777;
            DataStreamSource<String> data = env.socketTextStream(hostname, port);
     
            // 创建窗口（TimeWindow)
            // 1.MapFunction第一个泛型表示输入的数据类型；第二个泛型表示输出的数据类型（可以自定义）
            SingleOutputStreamOperator<Tuple1<String>> tupled = data.map(new MapFunction<String, Tuple1<String>>() {
     
                /**
                 * 键输入的字符串转成一元组
                 * @param s 输入的字符串
                 * @return 转化后的一元组
                 * @throws Exception
                 */
                public Tuple1<String> map(String s) throws Exception {
                    return new Tuple1<String>(s);
                }
            });
     
            // 2.按照一元组第一个元素分组（只有一个元素只能写0）
            KeyedStream<Tuple1<String>, Tuple> keyed = tupled.keyBy(0);
     
            // 3.获取时间滑动窗口
            // Time.seconds(10)表示时间窗口大小，Time.seconds(5)表示时间间隔多少开始滑动
            WindowedStream<Tuple1<String>, Tuple, TimeWindow> timeWindow = keyed.timeWindow(Time.seconds(10), Time.seconds(5));
     
            // 利用时间滚动窗口对一个窗口内的数据进行处理
            // WindowFunction第一个泛型就是输入数据的数据类型；第二个泛型就是输出数据的数据类型；
            // 第三个就是key的数据类型；第四个就是窗口的数据类型
            SingleOutputStreamOperator<String> result = timeWindow.apply(new WindowFunction<Tuple1<String>, String, Tuple, TimeWindow>() {
     
                /**
                 * 数据具体处理过程
                 * @param tuple key就是分组key
                 * @param timeWindow 时间窗口
                 * @param iterable key组内的成员
                 * @param collector 数据处理后返回的结果的对象
                 * @throws Exception
                 */
                public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<Tuple1<String>> iterable, Collector<String> collector) throws Exception {
                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                    // 窗口实际开始时间
                    long start = timeWindow.getStart();
                    // 窗口实际结束时间
                    long end = timeWindow.getEnd();
     
                    String startTime = simpleDateFormat.format(start);
                    String endTime = simpleDateFormat.format(end);
                    // 对key进行处理
                    String key = tuple.getField(0).toString();
     
                    System.out.println("key:" + key + "...startTime:" + startTime + "...endTime:" + endTime);
                    // 目的就是将一元组转成输入时的字符串
                    for (Tuple1<String> t : iterable) {
                        String s = t.getField(0).toString();
                        collector.collect(s);
                    }
                }
            });
            result.print();
            env.execute();
        }
    }

[b4ed7479b5714a439a798588d0f8e00f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/6c46e0f9afad4befa240dd79dd96667b.png