Flink 5 — Flink SQL & Table API

Vì sao cần SQL cho stream processing

Bốn bài đầu của series Flink đều xoay quanh DataStream API — viết Java/Scala, tự lắp watermark, keyBy, window, quản lý state. Sức mạnh thì đầy đủ, nhưng rào cản kỹ thuật rất cao: một data analyst nghiệp vụ ngân hàng biết SQL rất giỏi nhưng không viết được Java sẽ đứng ngoài cuộc chơi realtime.

Flink SQL & Table API hạ rào cản đó. Bạn diễn đạt logic xử lý luồng bằng đúng cú pháp SQL quen thuộc (SELECT ... GROUP BY ... JOIN), Flink dịch nó thành cùng một đồ thị toán tử streaming mà DataStream API tạo ra — cùng watermark, cùng state, cùng checkpoint. Table API là biến thể lập trình (Java/Python) của cùng bộ máy: table.groupBy(...).select(...). Hai giao diện này dùng chung optimizer và runtime, nên bài này nói "Flink SQL" là bao hàm cả hai.

Điểm mấu chốt cần nắm ngay từ đầu: một câu SQL trong Flink không chạy một lần rồi kết thúc như trong PostgreSQL. Nó là một truy vấn liên tục (continuous query) — chạy mãi, và mỗi khi dữ liệu nguồn thay đổi thì kết quả cũng được cập nhật. Để hiểu điều này, phải bắt đầu từ khái niệm nền tảng nhất: dynamic table.

Dynamic table & stream-table duality

Trong cơ sở dữ liệu truyền thống, một TABLE là tập bản ghi tĩnh tại một thời điểm. Flink mở rộng ý niệm đó thành dynamic table (bảng động): một bảng thay đổi liên tục theo thời gian khi các sự kiện mới chảy vào. Bạn có thể hình dung nó như một bảng mà ai đó liên tục INSERT/UPDATE/DELETE hàng vào — nhưng do dòng dữ liệu điều khiển, không phải do người dùng gõ lệnh.

Đây là biểu hiện của stream-table duality (tính đối ngẫu luồng–bảng): một luồng sự kiện và một bảng thay đổi theo thời gian là hai cách nhìn cùng một thứ.

• Stream → Table: gom (materialize) một luồng lại theo thời gian thì được một bảng đang thay đổi. Ví dụ luồng giao dịch, nếu ta duy trì "tổng số giao dịch mỗi tài khoản", chính là một bảng động cập nhật theo từng giao dịch tới.
• Table → Stream: quan sát mọi thay đổi của một bảng động thì được một luồng các bản ghi thay đổi — gọi là changelog (nhật ký thay đổi).

Sơ đồ trên là toàn bộ mô hình thực thi của Flink SQL: luồng nguồn → dynamic table nguồn → truy vấn liên tục → dynamic table kết quả → changelog → sink. Mọi câu SQL bạn viết đều nằm ở khâu "truy vấn liên tục" giữa hai bảng động.

Continuous query & changelog

Xét câu (minh hoạ, cú pháp Flink SQL):

-- Minh hoạ Flink SQL — KHÔNG chạy trong sandbox PostgreSQL
SELECT account_id, COUNT(*) AS cnt
FROM transactions
GROUP BY account_id;

Trong PostgreSQL, câu này quét bảng một lần và trả về kết quả cố định. Trong Flink, transactions là bảng động; câu truy vấn chạy vĩnh viễn. Khi giao dịch đầu tiên của tài khoản A tới, kết quả có (A, 1). Giao dịch thứ hai tới, kết quả phải đổi thành (A, 2) — nghĩa là dòng (A, 1) cũ phải bị thu hồi và thay bằng (A, 2). Flink diễn đạt việc này qua changelog với các loại thay đổi:

Từ đó có hai loại changelog stream cực kỳ quan trọng phải phân biệt:

• Append stream (luồng chỉ thêm): kết quả chỉ sinh ra bản ghi mới, không bao giờ sửa/xoá bản cũ. Ví dụ một câu SELECT lọc đơn thuần, hoặc window aggregation (mỗi cửa sổ chốt một lần rồi thôi). Ghi ra sink append-only (như Kafka topic thường) rất dễ.
• Upsert / retract stream (luồng cập nhật/thu hồi): kết quả có thể sửa lại (như GROUP BY không cửa sổ ở trên, hoặc regular join). Ghi ra sink phải theo khoá chính (upsert) — ví dụ bảng JDBC, Kafka upsert (upsert-kafka), hoặc sink hiểu retract.

Hiểu bảng kết quả của mình là append hay upsert quyết định chọn sink nào và cấu hình ra sao. Chọn sai (đẩy upsert stream vào sink append-only) → Flink báo lỗi ngay khi khởi động job.

Định nghĩa bảng: CREATE TABLE + connector + watermark

Trước khi truy vấn, phải khai báo bảng động ánh xạ tới nguồn/đích thật. Flink dùng CREATE TABLE với mệnh đề WITH (...) chỉ định connector. Đây là chỗ Flink SQL khác PostgreSQL rõ nhất: bảng không lưu trữ dữ liệu, nó chỉ là lớp mô tả (schema + connector) trỏ tới Kafka, JDBC, filesystem…

-- Minh hoạ Flink SQL DDL — KHÔNG chạy trong sandbox PostgreSQL
CREATE TABLE transactions (
    txn_id      STRING,
    account_id  STRING,
    amount      DECIMAL(18, 2),
    currency    STRING,
    event_time  TIMESTAMP(3),
    -- khai báo watermark ngay trong DDL:
    WATERMARK FOR event_time AS event_time - INTERVAL '30' SECOND
) WITH (
    'connector' = 'kafka',
    'topic'     = 'core.transactions',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format'    = 'json',
    'scan.startup.mode' = 'latest-offset'
);

Ba điểm cốt lõi trong DDL này:

1. Connector: 'connector' = 'kafka' nói bảng này đọc từ Kafka. Các connector phổ biến gồm kafka, upsert-kafka, jdbc (Postgres/MySQL/Oracle), filesystem, elasticsearch. Cùng một câu SQL có thể đọc Kafka và ghi JDBC chỉ bằng cách khai báo hai bảng khác connector.
2. Time attribute + WATERMARK trong DDL: mệnh đề WATERMARK FOR event_time AS ... biến cột event_time thành rowtime attribute (thuộc tính thời gian sự kiện) và định nghĩa chiến lược watermark ngay tại đây — tương đương forBoundedOutOfOrderness(30s) của DataStream API mà bài event-time & watermark đã trình bày. Không có time attribute thì không dùng được window trên event time.
3. Muốn dùng processing time, thêm cột tính toán: proc_time AS PROCTIME().

Truy vấn streaming: window TVF & aggregation

Window Table-Valued Function (TVF)

Cách hiện đại (Flink 1.13+) để làm window trong SQL là Window TVF — hàm bảng nhận một bảng và trả về bảng đó đã được gắn thêm cột window_start, window_end. Ba loại tương ứng với window của DataStream API:

Đếm giao dịch mỗi tài khoản theo cửa sổ 5 phút (minh hoạ, cú pháp Flink SQL):

-- Minh hoạ Flink SQL — KHÔNG chạy trong sandbox PostgreSQL
SELECT
    window_start,
    window_end,
    account_id,
    COUNT(*)      AS txn_count,
    SUM(amount)   AS total_amount
FROM TABLE(
    TUMBLE(TABLE transactions, DESCRIPTOR(event_time), INTERVAL '5' MINUTES)
)
GROUP BY window_start, window_end, account_id;

Vì đã có watermark từ DDL, Flink biết khi nào một cửa sổ "đủ dữ liệu" và chốt nó — hệt như logic watermark trong DataStream API, chỉ khác là bạn không phải viết Java. Kết quả window aggregation là append stream (mỗi cửa sổ phát một lần khi chốt), nên ghi ra Kafka topic thường rất thẳng.

CUMULATE đặc biệt hữu ích cho nghiệp vụ: ví dụ "tổng giá trị giao dịch luỹ kế từ đầu ngày, cập nhật mỗi 15 phút" — cùng window_start = 00:00, window_end nhảy 00:15, 00:30, 00:45… cho tới hết ngày. Dashboard giám sát hạn mức trong ngày dùng đúng mẫu này.

Continuous aggregation (không cửa sổ)

Nếu bỏ window, GROUP BY account_id cho continuous aggregation — cập nhật vô thời hạn, sinh upsert/retract stream. Đây là điểm khác biệt hành vi cần khắc cốt: window aggregation append và tự "quên" state cũ khi cửa sổ chốt; còn continuous aggregation giữ state mãi (mỗi key một dòng kết quả) và không ngừng cập nhật. Với luồng vô hạn, aggregation không cửa sổ có thể phình state — cần TTL (state time-to-live) để dọn key cũ, chủ đề của bài quản lý trạng thái.

JOIN trong streaming

JOIN là nơi Flink SQL phân hoá thành nhiều biến thể, mỗi loại có ngữ nghĩa và chi phí state khác nhau. Chọn nhầm loại join là lỗi phổ biến nhất khi mới làm streaming SQL.

Regular join

FROM a JOIN b ON a.k = b.k thông thường. Flink phải giữ state cả hai phía mãi mãi để khi một hàng mới bên A tới còn khớp được với hàng bên B tới từ trước. State phình vô hạn nếu không đặt TTL. Cho kết quả retract stream. Chỉ nên dùng khi cả hai bảng nhỏ hoặc có TTL hợp lý.

Interval join

Ràng buộc join trong một khoảng thời gian giữa hai luồng: b.event_time BETWEEN a.event_time - INTERVAL '10' MINUTE AND a.event_time. Vì có biên thời gian, Flink chỉ giữ state trong khung đó rồi dọn — state có giới hạn. Hợp cho "ghép giao dịch với xác thực OTP xảy ra trong vòng 10 phút".

Temporal join & lookup join — làm giàu dữ liệu

Đây là hai loại quan trọng nhất cho nghiệp vụ ngân hàng: làm giàu (enrich) mỗi giao dịch bằng dữ liệu tham chiếu như tỷ giá, hạng khách hàng, danh mục.

Temporal join ghép mỗi bản ghi bên trái với phiên bản đúng thời điểm của bảng phải (dựa trên event time). Bảng phải là versioned table (bảng có phiên bản theo thời gian, thường từ changelog/CDC). Ví dụ: quy đổi giao dịch USD sang VND theo tỷ giá tại thời điểm giao dịch, không phải tỷ giá hiện tại:

-- Minh hoạ Flink SQL — KHÔNG chạy trong sandbox PostgreSQL
SELECT
    t.txn_id,
    t.amount,
    t.currency,
    t.amount * r.rate AS amount_vnd
FROM transactions AS t
JOIN fx_rates FOR SYSTEM_TIME AS OF t.event_time AS r
    ON t.currency = r.currency;

Cú pháp FOR SYSTEM_TIME AS OF t.event_time là đặc trưng temporal join — "lấy dòng tỷ giá có hiệu lực tại t.event_time". Điều này cho kết quả đúng đắn khi replay: quy đổi lại giao dịch cũ vẫn dùng đúng tỷ giá lúc đó, không bị "nhiễm" tỷ giá mới.

Lookup join tra cứu bảng ngoài (JDBC, HBase, Redis) ngay lúc chạy cho mỗi hàng bên trái, dùng processing time: FOR SYSTEM_TIME AS OF t.proc_time. Khác temporal join ở chỗ nó gọi ra hệ ngoài theo thời gian thực (thường có cache/async để giảm tải), phù hợp khi bảng tham chiếu nằm sẵn trong một database vận hành và bạn chấp nhận "giá trị hiện tại" thay vì "giá trị đúng thời điểm sự kiện".

Tích hợp CDC: từ binlog tới dynamic table

Bảng tham chiếu như fx_rates, customers, accounts sống trong database vận hành (MySQL/Postgres/Oracle). Flink CDC cho phép đọc binlog/redo log của các DB này và đưa trực tiếp vào Flink dưới dạng dynamic table đã mang sẵn changelog (+I/-U/+U/-D) — mỗi UPDATE ở nguồn thành một cặp retract trong Flink.

-- Minh hoạ Flink SQL (Flink CDC) — KHÔNG chạy trong sandbox PostgreSQL
CREATE TABLE fx_rates (
    currency   STRING,
    rate       DECIMAL(18, 6),
    valid_from TIMESTAMP(3),
    PRIMARY KEY (currency) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname'  = 'mysql-ops',
    'database-name' = 'reference',
    'table-name'    = 'fx_rates'
);

Sức mạnh nằm ở chỗ: bảng CDC này chính là một versioned table có thể dùng làm phía phải của temporal join ở trên. Toàn bộ chuỗi "core banking đổi tỷ giá → binlog → Flink CDC → temporal join làm giàu giao dịch" chạy end-to-end với độ trễ dưới giây, không cần batch ETL. Đây là hiện thân cụ thể của kiến trúc mô tả trong CDC & dữ liệu thời gian thực, nay diễn đạt gọn bằng SQL.

MATCH_RECOGNIZE — CEP bằng SQL

Flink SQL còn hỗ trợ complex event processing (CEP — xử lý sự kiện phức hợp) ngay trong SQL qua MATCH_RECOGNIZE: phát hiện mẫu chuỗi (sequence pattern) trên một luồng đã sắp theo thời gian. Ví dụ khái quát: tìm mẫu "3 giao dịch tăng dần liên tiếp rồi một giao dịch giảm mạnh" — dấu hiệu bất thường.

-- Minh hoạ Flink SQL (MATCH_RECOGNIZE) — KHÔNG chạy trong sandbox PostgreSQL
SELECT account_id, start_ts, drop_amount
FROM transactions
MATCH_RECOGNIZE (
    PARTITION BY account_id
    ORDER BY event_time
    MEASURES FIRST(A.event_time) AS start_ts, C.amount AS drop_amount
    PATTERN (A+ C)
    DEFINE
        A AS A.amount > COALESCE(LAST(A.amount, 1), 0),
        C AS C.amount < 0.5 * LAST(A.amount)
) AS T;

Cú pháp PATTERN (A+ C) dùng ký hiệu kiểu regex trên dòng sự kiện. MATCH_RECOGNIZE mạnh cho luật chống gian lận dạng chuỗi (velocity, escalation) mà GROUP BY khó diễn đạt. Đây chỉ là giới thiệu — CEP là chủ đề sâu riêng.

Catalog — quản lý metadata bảng

Bảng khai báo bằng CREATE TABLE mặc định chỉ tồn tại trong phiên hiện tại. Catalog lưu metadata bảng/view/function bền vững và chia sẻ giữa các job. Flink có:

• GenericInMemoryCatalog: mặc định, mất khi tắt phiên.
• HiveCatalog: lưu vào Hive Metastore — bảng khai báo một lần, mọi job và mọi analyst dùng lại.
• JdbcCatalog, catalog cho Iceberg/Paden lakehouse…

Với đội ngân hàng, dùng HiveCatalog để định nghĩa sẵn transactions, fx_rates, customers một lần; analyst chỉ việc SELECT mà không phải lặp lại DDL kết nối Kafka/CDC. Đây là nền tảng để Flink SQL trở thành công cụ self-service thật sự.

Use case thực tế: giám sát velocity & làm giàu tỷ giá bằng SQL

Bối cảnh. Đội giám sát NCB cần một job realtime vừa (a) đếm số giao dịch mỗi tài khoản theo cửa sổ 5 phút để bắt velocity bất thường, vừa (b) quy đổi mọi giao dịch ngoại tệ sang VND theo tỷ giá đúng thời điểm để tính tổng giá trị theo VND. Trước đây làm bằng batch mỗi giờ trên kho, độ trễ 60 phút và tỷ giá dùng sai (lấy tỷ giá cuối ngày). Analyst nghiệp vụ không viết Java nên không tự làm realtime được.

Thiết kế bằng Flink SQL (không một dòng Java):

1. Bảng nguồn transactions: connector kafka, cột event_time, WATERMARK ... - INTERVAL '30' SECOND ngay trong DDL (bù POS gửi trễ 20–35 giây, đúng logic bài watermark).
2. Bảng tham chiếu fx_rates: connector mysql-cdc đọc binlog từ core banking, PRIMARY KEY (currency) → thành versioned table.
3. Làm giàu: temporal join transactions ... JOIN fx_rates FOR SYSTEM_TIME AS OF t.event_time → có amount_vnd đúng tỷ giá tại thời điểm giao dịch.
4. Velocity: TUMBLE(TABLE ..., DESCRIPTOR(event_time), INTERVAL '5' MINUTES) + GROUP BY window, account_id, HAVING COUNT(*) > 20 để phát cảnh báo. Kết quả append → ghi Kafka topic cảnh báo.
5. Luỹ kế trong ngày: CUMULATE bước 15 phút để dashboard hiện tổng giá trị VND luỹ kế theo tài khoản/chi nhánh.

Kết quả (minh hoạ tình huống). Độ trễ từ giao dịch phát sinh tới cảnh báo giảm từ ~60 phút (batch) xuống ~35 giây; tổng giá trị VND chính xác vì dùng đúng tỷ giá thời điểm thay vì tỷ giá cuối ngày; và quan trọng về mặt tổ chức: 2 analyst nghiệp vụ tự viết và bảo trì được job này bằng SQL, không phụ thuộc đội Java. Con số minh hoạ để làm rõ tác động, không phải benchmark chính thức.

Để đối chiếu cùng bài toán trên kho phân tích cột (batch), có thể xem ClickHouse tổng quan — nơi cùng phép đếm/tổng theo cửa sổ được làm sau khi dữ liệu đã hạ cánh, đánh đổi độ trễ lấy khả năng truy vấn tuỳ biến trên lịch sử dài.

Ghi nhớ

• Flink SQL = streaming bằng SQL quen thuộc; Table API là biến thể lập trình. Cùng optimizer + runtime với DataStream API — hạ rào cản cho analyst.
• Dynamic table: bảng thay đổi liên tục theo dòng sự kiện. Stream-table duality: luồng và bảng-đổi-theo-thời-gian là hai cách nhìn cùng một thứ.
• Continuous query: câu SQL chạy mãi, cập nhật kết quả theo dữ liệu — khác hẳn PostgreSQL chạy một lần.
• Changelog: +I/-U/+U/-D. Append stream (window aggregation, lọc) chỉ thêm; upsert/retract stream (GROUP BY không cửa sổ, regular join) có sửa/xoá → chọn sink phù hợp (upsert-kafka, JDBC theo khoá).
• CREATE TABLE + WITH(connector): bảng chỉ là lớp mô tả trỏ Kafka/JDBC/filesystem. WATERMARK khai trong DDL tạo time attribute; PROCTIME() cho processing time.
• Window TVF: TUMBLE (cố định), HOP (trượt), CUMULATE (luỹ kế trong ngày) — trả bảng có window_start/window_end.
• JOIN: regular (state vô hạn, cần TTL), interval (biên thời gian, state bị chặn), temporal (FOR SYSTEM_TIME AS OF event_time — tỷ giá đúng thời điểm, đúng khi replay), lookup (tra DB ngoài theo proc_time).
• Flink CDC biến binlog thành dynamic table versioned → dùng làm phía phải temporal join để làm giàu giao dịch end-to-end dưới giây.
• MATCH_RECOGNIZE = CEP bằng SQL, phát hiện mẫu chuỗi (chống gian lận velocity/escalation). Catalog (HiveCatalog) lưu định nghĩa bảng bền vững → self-service.
• Cú pháp Flink SQL KHÁC PostgreSQL (dynamic table, TVF, FOR SYSTEM_TIME, connector DDL) → mọi block Flink SQL ở đây là minh hoạ, không chạy trong sandbox read-only.