滑動窗口（Sliding Windows）與滾動窗口類似，滑動窗口的大小也是固定的。區(qū)別在于，窗口之間并不是首尾相接的，而是可以“錯開”一定的位置。如果看作一個窗口的運動，那么就像是向前小步“滑動”一樣。定義滑動窗口的參數(shù)有兩個：除去窗口大?。╳indow size）之外，還有一個滑動步長（window slide），代表窗口計算的頻率。

1.png

demo演示：

場景：接收通過socket發(fā)送過來的數(shù)據(jù)，定義一個1小時的時間窗口大小，每30秒滑動觸發(fā)運算一次

（1）準備一個實體對象，消息對象

package com.pojo;

import java.io.Serializable;

/**
 * Created by lj on 2022-07-05.
 */
public class WaterSensor implements Serializable {
    private String id;
    private long ts;
    private int vc;

    public WaterSensor(){

    }

    public WaterSensor(String id,long ts,int vc){
        this.id = id;
        this.ts = ts;
        this.vc = vc;
    }

    public int getVc() {
        return vc;
    }

    public void setVc(int vc) {
        this.vc = vc;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public long getTs() {
        return ts;
    }

    public void setTs(long ts) {
        this.ts = ts;
    }
}

（2）編寫socket代碼，模擬數(shù)據(jù)發(fā)送

package com.producers;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Random;

/**
 * Created by lj on 2022-07-05.
 */
public class Socket_Producer {
    public static void main(String[] args) throws IOException {

        try {
            ServerSocket ss = new ServerSocket(9999);
            System.out.println("啟動 server ....");
            Socket s = ss.accept();
            BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(s.getOutputStream()));
            String response = "java,1,2";

            //每 2s 發(fā)送一次消息
            int i = 0;
            Random r=new Random();   
            String[] lang = {"flink","spark","hadoop","hive","hbase","impala","presto","superset","nbi"};

            while(true){
                Thread.sleep(2000);
                response= lang[r.nextInt(lang.length)] + "," + i + "," + i+"\n";
                System.out.println(response);
                try{
                    bw.write(response);
                    bw.flush();
                    i++;
                }catch (Exception ex){
                    System.out.println(ex.getMessage());
                }

            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

（3）從socket端接收數(shù)據(jù)，并設置30秒觸發(fā)執(zhí)行一次窗口運算

package com.examples;

import com.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
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.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by lj on 2022-07-06.
 *
 *滑動窗口（Sliding Windows）與滾動窗口類似，滑動窗口的大小也是固定的。區(qū)別在于，窗口之間并不是首尾相接的，而是可以“錯開”一定的位置。
 * 如果看作一個窗口的運動，那么就像是向前小步“滑動”一樣。定義滑動窗口的參數(shù)有兩個：除去窗口大?。╳indow size）之外，
 * 還有一個滑動步長（window slide），代表窗口計算的頻率。
 */
public class Flink_Group_Window_Hop {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        DataStreamSource<String> streamSource = env.socketTextStream("127.0.0.1", 9999,"\n");
        SingleOutputStreamOperator<WaterSensor> waterDS = streamSource.map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String s) throws Exception {
                String[] split = s.split(",");
                return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
            }
        });

        // 將流轉化為表
        Table table = tableEnv.fromDataStream(waterDS,
                $("id"),
                $("ts"),
                $("vc"),
                $("pt").proctime());

        tableEnv.createTemporaryView("EventTable", table);

        Table result = tableEnv.sqlQuery(
                "SELECT " +
                        "id, " +                //window_start, window_end,
                        "COUNT(ts) ,SUM(ts)" +
                        "FROM TABLE( " +
                        "HOP( TABLE EventTable , " +
                        "DESCRIPTOR(pt), " +
                        "INTERVAL '30' SECOND, INTERVAL '1' HOURS)) " +
                        "GROUP BY id , window_start, window_end"
        );

//        tableEnv.toChangelogStream(result).print("count");
//        tableEnv.toDataStream(result).print("toDataStream");
//        tableEnv.toAppendStream(result, Row.class).print("toAppendStream");           //追加模式
        tableEnv.toRetractStream(result, Row.class).print("toRetractStream");       //縮進模式
        env.execute();
    }
}

（4）效果演示

2.png

3.png

（5）總結
滾動窗口和滑動窗口的區(qū)別在于一個沒有重疊部分，一個有重疊部分。