Flink Table API 翻滚窗口：时间属性与常见陷阱解析

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

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

public class FlinkEventTimeWindowExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 模拟数据流：包含 GroupingColumn (String) 和 EventTimestamp (String)
        DataStream<Row> stream = env.fromElements(
                Row.of("A", "2023-01-01 10:00:00"),
                Row.of("B", "2023-01-01 10:00:05"),
                Row.of("A", "2023-01-01 10:00:10"),
                Row.of("C", "2023-01-01 10:00:15"),
                Row.of("B", "2023-01-01 10:00:20"),
                Row.of("A", "2023-01-01 10:00:25"),
                Row.of("D", "2023-01-01 10:10:00"), // 跨越到下一个窗口
                Row.of("A", "2023-01-01 10:09:58") // 乱序事件，但仍在水印延迟内
        );

        // 通过 Schema 声明事件时间属性
        Table table = tEnv.fromDataStream(stream,
            Schema.newBuilder()
                .column("f0", DataTypes.STRING()).as("GroupingColumn") // 原始字段 f0 映射为 GroupingColumn
                .column("f1", DataTypes.STRING()).as("EventTimestampStr") // 原始字段 f1 映射为 EventTimestampStr
                .columnByExpression("EventTime", "TO_TIMESTAMP(EventTimestampStr, 'yyyy-MM-dd HH:mm:ss')") // 派生 TIMESTAMP 列
                .watermark("EventTime", "EventTime - INTERVAL '5' SECOND") // 将 EventTime 声明为 ROWTIME，并定义5秒延迟的水印
                .build()
        );

        // 打印 Table Schema 确认时间属性已正确定义
        System.out.println("Table Schema with EventTime:");
        table.printSchema();

        // 定义翻滚窗口，基于 EventTime
        Table result = table
            .window(Tumble.over("10.minutes").on($("EventTime")).as("w"))
            .groupBy($("w"), $("GroupingColumn"))
            .select(
                $("GroupingColumn"),
                $("w").start().as("window_start"),
                $("w").end().as("window_end"),
                $("GroupingColumn").count().as("count")
            );

        result.execute().print();
    }
}

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

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

public class FlinkProcessingTimeWindowExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 模拟数据流
        DataStream<Row> stream = env.fromElements(
                Row.of("A", 1),
                Row.of("B", 2),
                Row.of("A", 3)
        );

        // 通过 Schema 声明处理时间属性
        Table table = tEnv.fromDataStream(stream,
            Schema.newBuilder()
                .column("f0", DataTypes.STRING()).as("GroupingColumn")
                .column("f1", DataTypes.INT()).as("Value")
                .columnByExpression("proc_time_attr", "PROCTIME()") // 声明一个虚拟的处理时间属性列
                .build()
        );

        System.out.println("Table Schema with ProcessingTime:");
        table.printSchema();

        // 定义翻滚窗口，基于 proc_time_attr (处理时间)
        Table result = table
            .window(Tumble.over("10.seconds").on($("proc_time_attr")).as("w"))
            .groupBy($("w"), $("GroupingColumn"))
            .select(
                $("GroupingColumn"),
                $("w").start().as("window_start"),
                $("w").end().as("window_end"),
                $("GroupingColumn").count().as("count")
            );

        result.execute().print();
    }
}

tEnv.executeSql(
    "CREATE TABLE my_source_table (" +
    "   GroupingColumn STRING," +
    "   EventTimestampStr STRING," +
    "   EventTime AS TO_TIMESTAMP(EventTimestampStr, 'yyyy-MM-dd HH:mm:ss')," +
    "   WATERMARK FOR EventTime AS EventTime - INTERVAL '5' SECOND" + // 声明事件时间及水印
    ") WITH (" +
    "   'connector' = 'datagen'," +
    "   'rows-per-second' = '1'," +
    "   'fields.GroupingColumn.length' = '1'," +
    "   'fields.EventTimestampStr.expression' = 'CAST(CURRENT_TIMESTAMP AS STRING)'" + // 示例，实际应从源读取
    ")"
);

Table table = tEnv.from("my_source_table");

Table result = table
    .window(Tumble.over("10.minutes").on($("EventTime")).as("w"))
    .groupBy($("w"), $("GroupingColumn"))
    .select(
        $("GroupingColumn"),
        $("w").start().as("window_start"),
        $("w").end().as("window_end"),
        $("GroupingColumn").count().as("count")
    );

result.execute().print();

tEnv.executeSql(
    "CREATE TABLE my_source_table_proc (" +
    "   GroupingColumn STRING," +
    "   Value INT," +
    "   proc_time_attr AS PROCTIME()" + // 声明处理时间属性
    ") WITH (" +
    "   'connector' = 'datagen'," +
    "   'rows-per-second' = '1'," +
    "   'fields.GroupingColumn.length' = '1'," +
    "   'fields.Value.kind' = 'sequence'," +
    "   'fields.Value.start' = '1'," +
    "   'fields.Value.end' = '100'" +
    ")"
);

Table table = tEnv.from("my_source_table_proc");

Table result = table
    .window(Tumble.