Flink Table API 翻滚窗口实践：时间属性定义与常见问题解析

import org.apache.flink.streaming.api.datastream.DataStream;
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 FlinkTumblingWindowExample {

    public static void main(String[] args) throws Exception {
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tableEnv = StreamTableEnvironment.create(settings);

        // 假设有一个模拟的 DataStream，包含设备ID、CPU使用率和事件时间
        // 实际应用中，这可能是一个从 Kafka 或其他源读取的 DataStream
        DataStream<Row> streamCPUData = tableEnv.fromValues(
            DataTypes.ROW(
                DataTypes.FIELD("deviceId", DataTypes.STRING()),
                DataTypes.FIELD("cpuUsageAverage", DataTypes.DOUBLE()),
                DataTypes.FIELD("time_Insert_ts", DataTypes.TIMESTAMP(3))
            ),
            Row.of("device_A", 0.5, java.sql.Timestamp.valueOf("2023-01-01 10:00:00")),
            Row.of("device_B", 0.7, java.sql.Timestamp.valueOf("2023-01-01 10:00:05")),
            Row.of("device_A", 0.6, java.sql.Timestamp.valueOf("2023-01-01 10:00:10")),
            Row.of("device_C", 0.8, java.sql.Timestamp.valueOf("2023-01-01 10:00:15")),
            Row.of("device_B", 0.75, java.sql.Timestamp.valueOf("2023-01-01 10:00:20")),
            Row.of("device_A", 0.55, java.sql.Timestamp.valueOf("2023-01-01 10:00:25")),
            Row.of("device_A", 0.65, java.sql.Timestamp.valueOf("2023-01-01 10:10:00")), // 下一个窗口
            Row.of("device_B", 0.85, java.sql.Timestamp.valueOf("2023-01-01 10:10:05"))
        );

        // 关键步骤：在 Schema 中定义事件时间属性和水位线
        Table tableCpuData = tableEnv.fromDataStream(streamCPUData, Schema.newBuilder()
                .column("deviceId", DataTypes.STRING())
                .column("cpuUsageAverage", DataTypes.DOUBLE())
                // 声明 'time_Insert_ts' 为 TIMESTAMP(3) 类型
                .column("time_Insert_ts", DataTypes.TIMESTAMP(3))
                // 将 'time_Insert_ts' 注册为事件时间属性，并设置水位线策略
                // 此处表示水位线滞后 'time_Insert_ts' 5秒
                .watermark("time_Insert_ts", "time_Insert_ts - INTERVAL '5' SECOND")
                .build());

        // 打印表结构，确认时间属性是否正确注册
        tableCpuData.printSchema();

        // 创建翻滚窗口并聚合
        Table windowedTable = tableCpuData
            .window(Tumble.over("10.minutes").on($("time_Insert_ts").rowtime()).as("w")) // 使用事件时间属性
            .groupBy($("w"), $("deviceId"))
            .select(
                $("deviceId"),
                $("cpuUsageAverage").avg().as("avgCpuUsage"),
                $("w").start().as("windowStart"),
                $("w").end().as("windowEnd")
            );

        // 输出结果到控制台
        windowedTable.execute().print();
    }
}

(
  `deviceId` STRING,
  `cpuUsageAverage` DOUBLE,
  `time_Insert_ts` TIMESTAMP(3) *ROWTIME*
)

// 假设 table 来源于 SQL 查询，其中 EventTimestamp 是一个字符串
Table table = tEnv.sqlQuery("SELECT EventTimestamp, GroupingColumn FROM MySourceTable");

// 步骤1: 将字符串时间戳转换为 TIMESTAMP 类型并创建新列 EventTime
// 这一步本身是正确的，它创建了一个新的 TIMESTAMP 列
table = table.addColumns($("EventTimestamp").toTimestamp().as("EventTime"));

// 步骤2: 注册 EventTime 为时间属性
// 如果是事件时间，需要通过 TableDescriptor 或 SQL DDL 注册水位线
// 这里我们假设使用 SQL DDL 进行注册，或者将 Table 转换回 DataStream 再注册 Schema
// Flink Table API 的 addColumns 无法直接注册 ROWTIME/PROCTIME，
// 通常需要通过 TableDescriptor 或 SQL DDL 在 TableSource/TableSink 层面进行声明。

// 演示如何在 TableDescriptor 中定义时间属性（对于 Table Source）
// 假设我们有一个 Table Source，其 Schema 包含 EventTimestamp
tableEnv.createTemporaryTable("MyInputTable", TableDescriptor.forConnector("...")
    .schema(Schema.newBuilder()
        .column("EventTimestamp", DataTypes.STRING())
        .column("GroupingColumn", DataTypes.STRING())
        .columnByExpression("EventTime", "CAST(EventTimestamp AS TIMESTAMP(3))") // 计算列
        .watermark("EventTime", "EventTime - INTERVAL '5' SECOND") // 注册事件时间及水位线
        .build())
    .build());

Table inputTable = tableEnv.from("MyInputTable");

// 现在 inputTable 中的 EventTime 已经被注册为事件时间属性
Table windowedTable = inputTable
    .window(Tumble.over("10.minutes").on($("EventTime").rowtime()).as("w"))
    .groupBy($("w"), $("GroupingColumn"))
    .select(
        $("GroupingColumn"),
        $("w").start().as("windowStart"),
        $("w").end().as("windowEnd"),
        $("GroupingColumn").count().as("count")
    );

windowedTable.execute().print();

// 假设 table 来源于 SQL 查询
Table table = tEnv.sqlQuery("SELECT EventTimestamp, GroupingColumn FROM MySourceTable");

// 添加一个处理时间列 'proc_time'
table = table.addColumns($("PROCTIME").as("proc_time"));

// 使用处理时间创建翻滚窗口
Table windowedTable = table
    .window(Tumble.over("10.minutes").on($("proc_time").proctime()).as("w"))
    .groupBy($("w"), $("GroupingColumn"))
    .select(
        $("GroupingColumn"),
        $("w").start().as("windowStart"),
        $("w").end().as("windowEnd"),
        $("GroupingColumn").count().as("count")
    );

windowedTable.execute().print();