使用表API或SQL实现Apache Flink翻滚窗口时间偏移

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class RollingWindowExample {
    public static void main(String[] args) throws Exception {
        // 创建流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        // 定义输入流的表结构
        tEnv.executeSql("CREATE TABLE input_table (id INT, timestamp TIMESTAMP, value DOUBLE) " +
                "WITH ('connector' = 'kafka', 'topic' = 'input_topic', 'properties.bootstrap.servers' = 'localhost:9092')");

        // 定义输出流的表结构
        tEnv.executeSql("CREATE TABLE output_table (window_start TIMESTAMP, window_end TIMESTAMP, sum_value DOUBLE) " +
                "WITH ('connector' = 'kafka', 'topic' = 'output_topic', 'properties.bootstrap.servers' = 'localhost:9092')");

        // 执行翻滚窗口计算
        tEnv.executeSql("INSERT INTO output_table " +
                "SELECT TUMBLE_START(timestamp, INTERVAL '1' HOUR), TUMBLE_END(timestamp, INTERVAL '1' HOUR), SUM(value) " +
                "FROM input_table " +
                "GROUP BY TUMBLE(timestamp, INTERVAL '1' HOUR)");

        // 提交作业并执行
        env.execute();
    }
}