6.3 Flink Table API 与 SQL（概述）

// DataStream API：20 行代码
stream.keyBy(Event::getSymbol)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .aggregate(new CountAggregate());

// SQL：1 行
SELECT symbol, COUNT(*) FROM events 
GROUP BY symbol, TUMBLE(event_time, INTERVAL '1' MINUTE)

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 注册 Kafka 表
String createTable = """
    CREATE TABLE events (
        eventId STRING,
        symbol STRING,
        amount DOUBLE,
        -- v3 的 eventTime 是毫秒时间戳（JSON 里是数字），先按 BIGINT 读，
        -- 再算出一个 TIMESTAMP_LTZ 列给 Watermark 用
        eventTime BIGINT,
        event_time AS TO_TIMESTAMP_LTZ(eventTime, 3),
        WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'eventd-cleaned',
        'properties.bootstrap.servers' = 'kafka:9092',
        'properties.group.id' = 'eventd-sql',   -- Source 表必填
        'format' = 'json'
    )
    """;
tableEnv.executeSql(createTable);

// 滚动窗口聚合（1.20 推荐的窗口 TVF 写法）
Table result = tableEnv.sqlQuery("""
    SELECT
        window_start,
        window_end,
        symbol,
        COUNT(*) as order_count,
        SUM(amount) as total_amount
    FROM TABLE(
        TUMBLE(TABLE events, DESCRIPTOR(event_time), INTERVAL '1' MINUTE))
    GROUP BY window_start, window_end, symbol
    """);

// DataStream → Table
DataStream<Event> stream = ...;
Table table = tableEnv.fromDataStream(
    stream,
    $("eventId"), $("symbol"), $("amount"), $("eventTime").rowtime());

// Table → DataStream：只接受 add-only 结果（窗口聚合属于这种）
DataStream<Row> resultStream = tableEnv.toDataStream(result);

// Table → 通用变更流：普通 GROUP BY 这类会更新结果、可能撤回旧行的查询必须用这个
DataStream<Row> changelogStream = tableEnv.toChangelogStream(result);

// 1. DataStream 读取 Kafka
DataStream<Event> stream = env.fromSource(kafkaSource, watermarkStrategy, "events");

// 2. 转为 Table 用 SQL 聚合
Table table = tableEnv.fromDataStream(stream, ...);
tableEnv.createTemporaryView("events", table);
Table aggregated = tableEnv.sqlQuery("SELECT ... FROM events GROUP BY ...");

// 3. 转回 DataStream 做 CEP
DataStream<Row> aggregatedStream = tableEnv.toDataStream(aggregated);
Pattern<Row, ?> pattern = Pattern.<Row>begin("start").where(...);
CEP.pattern(aggregatedStream, pattern).select(...);