Flink SQL과 Table API

왜 SQL인가 — DataStream API 위의 관계형 추상화

3장까지 다룬 DataStream API는 강력하지만 진입장벽이 있다. 상태 선언, 워터마크 전략, 윈도 컴포넌트를 직접 조립해야 한다. Flink SQL과 Table API는 그 위에 관계형 추상화를 얹어, 스트리밍 처리를 SQL 문장이나 메서드 체인으로 표현할 수 있게 한다.

두 API는 같은 옵티마이저(Calcite 기반 Blink Planner)를 공유하며, 동일한 실행 계획으로 컴파일된다.

세 API는 혼합해서 쓸 수 있다. Table을 DataStream으로, DataStream을 Table로 변환하는 브리지(bridge)를 Flink가 제공한다.

TableEnvironment — 통합 진입점

모든 Table / SQL 작업은 TableEnvironment에서 시작한다. 스트리밍이면 StreamTableEnvironment, 배치이면 BatchTableEnvironment를 사용한다. Flink 1.12부터는 StreamTableEnvironment 하나로 스트리밍과 배치를 모두 처리한다.

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

// SQL로 테이블 생성 (Kafka 커넥터)
tableEnv.executeSql("""
    CREATE TABLE user_events (
        user_id  BIGINT,
        event    STRING,
        ts       TIMESTAMP(3),
        WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic'     = 'user-events',
        'properties.bootstrap.servers' = 'kafka:9092',
        'format'    = 'json'
    )
""");

// SQL로 집계
TableResult result = tableEnv.executeSql("""
    SELECT user_id, COUNT(*) AS cnt
    FROM   user_events
    GROUP  BY user_id, TUMBLE(ts, INTERVAL '1' MINUTE)
""");

WATERMARK FOR ts AS ts - INTERVAL '5' SECOND는 DDL 안에서 이벤트 시간 컬럼과 워터마크 전략을 선언한다. DataStream API에서 WatermarkStrategy를 코드로 지정하는 것과 동일한 역할이다.

동적 테이블(Dynamic Tables)과 변경 로그 스트림(Changelog Stream)

Flink SQL의 핵심 개념은 동적 테이블(Dynamic Table) 이다. 일반 RDBMS 테이블은 어느 시점에 쿼리해도 같은 행 집합을 반환한다. 동적 테이블은 스트리밍 데이터와 함께 지속적으로 변화 한다.

GROUP BY 같은 집계 쿼리는 출력을 갱신(update)해야 하므로 변경 로그 스트림(changelog stream)을 생성한다. 반면 SELECT + filter + projection만 있는 쿼리는 append-only 스트림으로 출력된다.

Window TVF: SQL에서 윈도 처리

Flink SQL은 Window Table-Valued Function(TVF) 으로 윈도 집계를 표현한다. Flink 1.13에 도입된 방식이며, 기존 GROUP BY TUMBLE(...) 문법보다 유연하다.

-- TUMBLE: 겹침 없는 5분 윈도
SELECT
    window_start, window_end,
    user_id,
    COUNT(*) AS event_count
FROM TABLE(
    TUMBLE(TABLE user_events, DESCRIPTOR(ts), INTERVAL '5' MINUTES)
)
GROUP BY window_start, window_end, user_id;

-- HOP: 1분마다 슬라이드되는 5분 윈도
SELECT window_start, window_end, user_id, COUNT(*) AS cnt
FROM TABLE(
    HOP(TABLE user_events, DESCRIPTOR(ts),
        INTERVAL '1' MINUTES,   -- 슬라이드 간격
        INTERVAL '5' MINUTES)   -- 윈도 크기
)
GROUP BY window_start, window_end, user_id;

-- SESSION: 30초 이상 비활성이면 새 세션
SELECT window_start, window_end, user_id, COUNT(*) AS cnt
FROM TABLE(
    SESSION(TABLE user_events, DESCRIPTOR(ts),
            INTERVAL '30' SECONDS)
)
GROUP BY window_start, window_end, user_id;

Window TVF는 window_start, window_end, window_time 세 컬럼을 자동으로 추가한다. window_time은 타임 어트리뷰트이므로 이 컬럼으로 후속 윈도 연산을 체인할 수 있다.

Temporal Join: 서서히 변하는 차원 테이블과 조인

실시간 주문 스트림에 상품 가격 테이블을 조인할 때, 가격이 변하면 주문 당시의 가격을 써야 한다. 이것이 Temporal Join 이다.

-- 버전 관리 테이블 (FOR SYSTEM_TIME AS OF)
SELECT
    o.order_id,
    o.product_id,
    o.order_ts,
    p.price                  -- 주문 시점의 가격
FROM orders AS o
LEFT JOIN product_prices FOR SYSTEM_TIME AS OF o.order_ts AS p
    ON o.product_id = p.product_id;

FOR SYSTEM_TIME AS OF o.order_ts는 SQL:2011 표준 문법이다. Flink는 product_prices 테이블에서 o.order_ts 시각에 유효했던 행을 찾아 조인한다. product_prices는 주 키(primary key)와 이벤트 시간 컬럼이 있어야 한다.

Table API — 타입 안전한 관계형 연산

Java/Python 개발자라면 문자열 SQL 대신 메서드 체인으로 같은 로직을 표현할 수 있다.

Table userEvents = tableEnv.from("user_events");

Table result = userEvents
    .filter($("amount").isGreater(100))
    .groupBy($("user_id"),
             call("TUMBLE", $("ts"), lit(5).minutes()).as("w"))
    .select(
        $("user_id"),
        $("amount").sum().as("total"),
        call("TUMBLE_START", $("ts"), lit(5).minutes()).as("window_start"));

tableEnv.toDataStream(result).print();

Table API 표현식은 컴파일 시 타입이 검증된다. 복잡한 로직은 DataStream API에서 처리하고, 결과를 다시 Table로 바꿔 SQL과 조합할 수도 있다.

DataStream ↔ Table 브리지

// DataStream → Table (이벤트 시간 어트리뷰트 지정)
Table table = tableEnv.fromDataStream(
    orderStream,
    Schema.newBuilder()
        .column("order_id", DataTypes.BIGINT())
        .column("amount",   DataTypes.DOUBLE())
        .columnByExpression("ts", "CAST(event_ts AS TIMESTAMP_LTZ(3))")
        .watermark("ts", "ts - INTERVAL '5' SECOND")
        .build());

// Table → DataStream (append-only일 때)
DataStream<Row> outStream = tableEnv.toDataStream(table);

// Table → DataStream (업데이트 포함, changelog)
DataStream<Row> changelogStream = tableEnv.toChangelogStream(table);

카탈로그와 커넥터

카탈로그는 테이블·뷰·함수 메타데이터를 저장하는 레지스트리다. 기본값은 in-memory 카탈로그이며, Hive Metastore를 카탈로그로 등록하면 Hive 테이블을 Flink SQL에서 그대로 쿼리할 수 있다.

커넥터는 DDL의 WITH 절로 지정한다. 주요 커넥터:

DataStream API와 언제 무엇을 선택하는가

Flink SQL / Table API가 간결하지만 DataStream API를 써야 할 때도 있다.

References

• https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/tableapi/
• https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/concepts/dynamic_tables/
• https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/table/sql/queries/window-tvf/
• https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/queries/joins/
• https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/common/
• https://developer.confluent.io/courses/apache-flink/flink-sql/
• https://www.confluent.io/blog/getting-started-with-apache-flink-table-api/
• https://streamkap.com/resources-and-guides/flink-sql-tumbling-windows