Kafka 6 — Kafka Streams & ksqlDB

Vì sao cần xử lý luồng (stream processing)?

Trong kiến trúc dữ liệu truyền thống, chúng ta thu thập sự kiện, đổ vào kho lưu trữ, rồi định kỳ chạy các job batch (mỗi giờ, mỗi ngày) để biến đổi và tổng hợp. Cách này có độ trễ cố hữu: khi báo cáo sẵn sàng thì dữ liệu đã cũ hàng giờ. Với nhiều bài toán ngân hàng — phát hiện gian lận thẻ, giám sát hạn mức, cảnh báo giao dịch bất thường — độ trễ đó là không chấp nhận được. Kẻ gian đã rút tiền xong trước khi job batch buổi tối kịp phát hiện.

Xử lý luồng đảo ngược mô hình: thay vì "lưu trước, tính sau", ta biến đổi và tổng hợp dữ liệu ngay khi sự kiện tới, liên tục, không có điểm kết thúc. Mỗi giao dịch mới đi qua topology xử lý trong vài mili-giây và cập nhật kết quả tức thì. Kafka là nền tảng lý tưởng cho việc này vì nó vốn đã là một log sự kiện có thứ tự, bền vững, có thể phát lại (replay) — đúng thứ mà một máy xử lý luồng cần để đảm bảo tính đúng đắn khi khôi phục sau sự cố.

Có hai công cụ chính thức trong hệ sinh thái Kafka để làm việc này: Kafka Streams (thư viện Java) và ksqlDB (lớp SQL đặt trên Kafka Streams).

Kafka Streams là gì?

Kafka Streams không phải là một cluster xử lý riêng biệt như Spark hay Flink. Nó là một thư viện client (org.apache.kafka:kafka-streams) mà bạn nhúng thẳng vào ứng dụng Java/Scala của mình. Ứng dụng của bạn vẫn là một tiến trình JVM thông thường — bạn triển khai nó như bất kỳ microservice nào (JAR, container, Kubernetes pod), không cần dựng thêm hạ tầng resource manager nào cả.

Điều này có vài hệ quả quan trọng:

• Scale bằng partition. Song song hóa của Kafka Streams gắn chặt với số partition của topic đầu vào. Nếu topic có 12 partition, bạn có thể chạy tối đa 12 luồng xử lý (stream thread) đồng thời — chia trên bao nhiêu instance ứng dụng cũng được. Thêm instance thứ hai, group coordinator sẽ rebalance partition giữa các instance, y hệt như consumer group. Muốn tăng song song hơn nữa thì phải tăng số partition.
• Chịu lỗi tự động. Nếu một instance chết, các partition (và cả state store của chúng) được phân phối lại cho instance còn sống. Không cần master node điều phối.
• Đơn vị tính là application.id. Mọi instance dùng chung một application.id tạo thành một nhóm cộng tác, chia sẻ tải và chia sẻ state.

Streams cung cấp hai tầng API: DSL (Domain Specific Language — map, filter, groupBy, aggregate, join...) cho phần lớn nhu cầu, và Processor API cấp thấp khi bạn cần điều khiển thủ công state store, punctuation theo thời gian, hoặc topology tùy biến.

Nhị nguyên Stream–Table: KStream vs KTable

Đây là khái niệm nền tảng nhất, và cũng dễ nhầm nhất, của Kafka Streams.

KStream — chuỗi sự kiện

KStream biểu diễn một chuỗi bản ghi độc lập, chỉ-thêm (append-only). Mỗi bản ghi là một sự kiện đã xảy ra — một sự thật bất biến. Hai bản ghi cùng key không ghi đè lẫn nhau; chúng là hai sự kiện khác nhau.

Ví dụ, luồng giao dịch:

(acc-01, +100)
(acc-01, -30)
(acc-01, +50)

Đây là ba giao dịch riêng biệt trên tài khoản acc-01, không phải ba lần cập nhật một giá trị.

KTable — trạng thái mới nhất theo key

KTable biểu diễn trạng thái hiện tại: với mỗi key, chỉ giữ giá trị mới nhất. Bản ghi cùng key được hiểu là một bản cập nhật (upsert); giá trị null là một tombstone (xóa key khỏi bảng).

Cùng dữ liệu như trên, nếu diễn giải là KTable thì acc-01 cuối cùng có giá trị +50 (giá trị mới nhất ghi đè các giá trị cũ).

Vì sao gọi là "nhị nguyên"?

Stream và table là hai mặt của cùng một dữ liệu:

• Table → Stream (changelog): mỗi thay đổi trạng thái của một bảng có thể được phát ra thành một dòng sự kiện thay đổi. Đây chính là changelog.
• Stream → Table: nếu bạn "phát lại" changelog và liên tục upsert vào một bản đồ theo key, bạn dựng lại được bảng.

Mối liên hệ này ánh xạ trực tiếp sang Kafka: một topic có cleanup policy = compact chính là một changelog. Log compaction giữ lại bản ghi mới nhất cho mỗi key và dọn các bản cũ, đúng ngữ nghĩa của một bảng. Đó là lý do state store của Kafka Streams được sao lưu bằng một changelog topic dạng compacted — bảng trong bộ nhớ và log trên Kafka là cùng một thứ nhìn từ hai góc.

Ngoài ra còn có GlobalKTable: một bảng được nhân bản toàn bộ trên mọi instance (không phân mảnh theo partition), rất tiện cho join tra cứu dữ liệu tham chiếu nhỏ (bảng khách hàng, bảng loại thẻ...).

Toán tử stateless vs stateful

Stateless — không cần nhớ gì

Các toán tử này xử lý từng bản ghi độc lập, không lưu trạng thái giữa các bản ghi:

• mapValues / map — biến đổi giá trị (hoặc cả key).
• filter — giữ lại bản ghi thỏa điều kiện.
• flatMap — một bản ghi vào, nhiều bản ghi ra.
• branch (nay là split().branch()) — chia một luồng thành nhiều luồng theo điều kiện.
• foreach, peek — tác dụng phụ.

Vì không có trạng thái, chúng nhẹ, không cần state store, và song song hóa hoàn toàn tự do.

Stateful — cần bộ nhớ trạng thái

Khi bạn count, reduce, aggregate hay join, máy xử lý phải nhớ kết quả trung gian giữa các bản ghi. Trạng thái này được giữ trong một state store.

• State store mặc định là RocksDB — một key-value store nhúng, lưu trên đĩa cục bộ của instance. Việc dùng đĩa cho phép state lớn hơn RAM mà không tràn heap JVM.
• Sao lưu bằng changelog topic. Mọi thay đổi state được ghi song song vào một changelog topic (compacted) trên Kafka. Khi instance chết và partition được chuyển sang instance khác, state store được phục hồi bằng cách phát lại changelog. Đây là cơ chế đảm bảo state không mất khi có sự cố.
• Standby replica (tùy chọn num.standby.replicas) giữ sẵn bản sao state ở instance khác để rút ngắn thời gian phục hồi.

Một toán tử stateful điển hình luôn bắt đầu bằng groupByKey (hoặc groupBy nếu cần đổi key), rồi mới aggregate/count/reduce. Kết quả của aggregate là một KTable.

Windowing — cửa sổ thời gian

Nhiều phép tổng hợp chỉ có ý nghĩa trong một khoảng thời gian ("số giao dịch mỗi tài khoản trong 5 phút gần nhất"). Windowing chia dòng vô hạn thành các cửa sổ hữu hạn để tổng hợp.

Dữ liệu đến trễ và grace

Trong thực tế, sự kiện có thể đến muộn (mạng chậm, thiết bị offline rồi đồng bộ lại). Kafka Streams phân biệt event time (thời điểm sự kiện thực sự xảy ra, lấy từ timestamp trong bản ghi) với processing time (thời điểm máy nhận được nó).

Với windowing, bạn khai báo một grace period: khoảng thời gian tiếp tục chấp nhận bản ghi đến trễ sau khi cửa sổ đã kết thúc theo event time. Bản ghi đến sau grace period bị coi là quá muộn và bị bỏ. Cấu hình grace hợp lý là sự đánh đổi giữa độ chính xác (grace dài, chờ dữ liệu trễ) và độ trễ kết quả cùng chi phí giữ state (grace ngắn, chốt sổ sớm).

Joins — kết hợp nhiều luồng

Kafka Streams hỗ trợ ba dạng join, mỗi dạng có ngữ nghĩa khác nhau:

• Stream–Stream join — kết hợp hai KStream. Bắt buộc có cửa sổ (windowed), vì hai sự kiện chỉ join được khi xảy ra "gần nhau" về thời gian. Ví dụ: khớp sự kiện "quẹt thẻ" với sự kiện "xác thực OTP" trong vòng 30 giây. Hỗ trợ inner/left/outer join.
• Stream–Table join — làm giàu (enrich) mỗi sự kiện trong stream bằng trạng thái mới nhất của một table. Không cần cửa sổ; đây là "lookup". Ví dụ: mỗi giao dịch tra bảng khách hàng để gắn thông tin hồ sơ. Hỗ trợ inner/left.
• Table–Table join — kết hợp trạng thái mới nhất của hai bảng thành một bảng mới; bất kỳ thay đổi ở hai bên đều cập nhật kết quả. Hỗ trợ inner/left/outer.

Điều kiện quan trọng cho join phân mảnh (không dùng GlobalKTable): hai bên phải co-partitioned — cùng số partition và cùng chiến lược phân vùng theo key, để các bản ghi cùng key rơi vào cùng partition và cùng một stream task xử lý được.

Sơ đồ topology

Exactly-once processing

Mặc định, xử lý luồng chỉ đảm bảo at-least-once: khi có sự cố và consumer phát lại, một sự kiện có thể được xử lý hai lần, khiến phép count bị đếm trùng. Với bài toán tài chính, đó là lỗi nghiêm trọng.

Kafka Streams hỗ trợ exactly-once semantics (EOS) thông qua một cấu hình duy nhất:

processing.guarantee = exactly_once_v2

Cơ chế đằng sau dựa trên giao dịch Kafka (transactions) và idempotent producer: việc đọc offset đầu vào, cập nhật state store (qua changelog) và ghi kết quả đầu ra được gói trong một giao dịch nguyên tử. Hoặc tất cả cùng commit, hoặc không có gì được thấy bởi consumer đọc với isolation.level=read_committed. Nhờ đó, dù có crash và phát lại, mỗi sự kiện chỉ ảnh hưởng đến kết quả đúng một lần.

Lưu ý: exactly_once_v2 (từ Kafka 2.5, thay cho exactly_once cũ) hiệu quả hơn nhiều vì dùng chung một producer cho tất cả partition thay vì một producer cho mỗi cặp task, giảm chi phí đáng kể. EOS chỉ đảm bảo trong phạm vi Kafka; nếu bạn ghi ra hệ thống ngoài (DB, API) thì phải tự lo idempotency ở đó.

ksqlDB — stream processing bằng SQL

ksqlDB là một lớp trên Kafka Streams cho phép viết logic xử lý luồng bằng SQL thay vì code Java. Bạn khai báo STREAM và TABLE (ánh xạ đúng nhị nguyên KStream/KTable), rồi viết các câu truy vấn liên tục.

Lưu ý: đây là ngôn ngữ SQL của ksqlDB, chạy trên ksqlDB server, không phải một "SQL Builder" hay SQL của cơ sở dữ liệu quan hệ. Cú pháp giống SQL nhưng ngữ nghĩa là truy vấn liên tục trên dòng sự kiện.

Có hai loại truy vấn:

• Push query (EMIT CHANGES): chạy vô thời hạn, đẩy kết quả mới mỗi khi có sự kiện mới. Đây là bản chất stream processing.
• Pull query: truy vấn điểm một lần vào trạng thái hiện tại của một table thực thể hóa (materialized), giống query cơ sở dữ liệu thông thường — trả về rồi kết thúc.

Ví dụ mã

Kafka Streams (Java) — đếm giao dịch theo tài khoản trong cửa sổ 5 phút, kèm phát hiện ngưỡng

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "fraud-detector");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
// Bật exactly-once cho chính xác tài chính
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
          StreamsConfig.EXACTLY_ONCE_V2);

StreamsBuilder builder = new StreamsBuilder();

// Stream giao dịch: key = accountId, value = Transaction
KStream<String, Transaction> tx =
    builder.stream("transactions",
        Consumed.with(Serdes.String(), transactionSerde));

// Stateless: chỉ giữ giao dịch hợp lệ
KStream<String, Transaction> valid =
    tx.filter((accId, t) -> t.getAmount() > 0);

// Stateful: đếm số giao dịch mỗi tài khoản trong cửa sổ tumbling 5 phút,
// cho phép dữ liệu trễ 1 phút (grace)
KTable<Windowed<String>, Long> counts = valid
    .groupByKey(Grouped.with(Serdes.String(), transactionSerde))
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5), Duration.ofMinutes(1)))
    .count(Materialized.as("tx-counts-5m")); // state store RocksDB

// Phát hiện gian lận: >10 giao dịch/5 phút -> cảnh báo
counts.toStream()
    .filter((windowedKey, cnt) -> cnt != null && cnt > 10)
    .map((windowedKey, cnt) -> KeyValue.pair(
        windowedKey.key(),
        new Alert(windowedKey.key(), cnt, windowedKey.window().start())))
    .to("fraud-alerts",
        Produced.with(Serdes.String(), alertSerde));

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

Làm giàu bằng stream–table join (Java)

// Bảng khách hàng: trạng thái mới nhất theo customerId
KTable<String, Customer> customers =
    builder.table("customers",
        Consumed.with(Serdes.String(), customerSerde));

// Đổi key giao dịch sang customerId rồi join lookup (không cần cửa sổ)
KStream<String, EnrichedTx> enriched = tx
    .selectKey((accId, t) -> t.getCustomerId())
    .join(customers,
          (t, c) -> new EnrichedTx(t, c.getSegment(), c.getRiskLevel()),
          Joined.with(Serdes.String(), transactionSerde, customerSerde));

enriched.to("enriched-transactions");

ksqlDB — cùng logic bằng SQL

-- ksqlDB: khai báo STREAM trên topic giao dịch
CREATE STREAM transactions (
    account_id  VARCHAR KEY,
    customer_id VARCHAR,
    amount      DECIMAL(18,2),
    ts          BIGINT
) WITH (
    KAFKA_TOPIC  = 'transactions',
    VALUE_FORMAT = 'JSON',
    TIMESTAMP    = 'ts'
);

-- ksqlDB: TABLE làm giàu (trạng thái mới nhất theo customer_id)
CREATE TABLE customers (
    customer_id VARCHAR PRIMARY KEY,
    segment     VARCHAR,
    risk_level  VARCHAR
) WITH (
    KAFKA_TOPIC  = 'customers',
    VALUE_FORMAT = 'JSON'
);

-- ksqlDB: tổng hợp có cửa sổ + phát hiện gian lận,
-- tạo một STREAM kết quả (persistent query chạy liên tục)
CREATE TABLE fraud_alerts AS
    SELECT account_id,
           COUNT(*)          AS tx_count,
           WINDOWSTART       AS window_start
    FROM   transactions
    WINDOW TUMBLING (SIZE 5 MINUTES, GRACE PERIOD 1 MINUTE)
    WHERE  amount > 0
    GROUP  BY account_id
    HAVING COUNT(*) > 10
    EMIT CHANGES;

-- ksqlDB: push query để quan sát cảnh báo theo thời gian thực
SELECT * FROM fraud_alerts EMIT CHANGES;

Khi nào dùng Streams, Connect hay Flink?

Ba công cụ trả lời ba câu hỏi khác nhau; đừng chọn nhầm.

• Kafka Connect — dùng khi việc cần làm là di chuyển dữ liệu giữa Kafka và hệ thống ngoài (DB, S3, Elasticsearch) mà không (hoặc rất ít) biến đổi logic. Connect là ETL nhập/xuất khai báo bằng cấu hình, không viết code. Không phải chỗ để làm aggregate hay join phức tạp.
• Kafka Streams / ksqlDB — dùng khi cả nguồn và đích đều trong Kafka và bạn cần biến đổi, tổng hợp, join theo trạng thái. Nhẹ (chỉ là thư viện/microservice), gắn chặt hệ sinh thái Kafka, phù hợp phần lớn nhu cầu stream processing "Kafka-to-Kafka". ksqlDB nếu team thiên về SQL; Streams nếu cần kiểm soát bằng code JVM.
• Apache Flink — dùng khi bài toán vượt ngoài Kafka: cần nhiều nguồn/đích không phải Kafka, cần quản lý state cực lớn, cần các đảm bảo và cửa sổ tinh vi, hoặc muốn một cluster xử lý dùng chung cho nhiều pipeline với savepoint/checkpoint mạnh. Flink là một hệ thống xử lý phân tán độc lập (có JobManager/TaskManager riêng), mạnh và linh hoạt hơn nhưng vận hành nặng hơn Kafka Streams.

Quy tắc thực dụng: nếu dữ liệu vào và ra đều nằm trong Kafka và team quen JVM/SQL, hãy bắt đầu bằng Kafka Streams hoặc ksqlDB. Chỉ nâng lên Flink khi thật sự chạm giới hạn.

Tóm tắt

• Xử lý luồng biến đổi và tổng hợp dữ liệu ngay khi sự kiện tới, thay cho batch có độ trễ cao — thiết yếu cho phát hiện gian lận và giám sát realtime.
• Kafka Streams là thư viện client nhúng vào ứng dụng JVM, không cần cluster riêng, scale theo số partition và rebalance như consumer group.
• KStream là chuỗi sự kiện append-only; KTable là trạng thái mới nhất theo key. Chúng là nhị nguyên của nhau: table ↔ changelog, gắn trực tiếp với topic compacted.
• Toán tử stateless (map/filter/branch) không cần bộ nhớ; stateful (aggregate/count/reduce/join) dùng state store RocksDB sao lưu bằng changelog topic.
• Windowing (tumbling, hopping, sliding, session) chia dòng vô hạn thành cửa sổ; grace period xử lý dữ liệu đến trễ theo event time.
• Joins: stream–stream (có cửa sổ), stream–table (lookup làm giàu), table–table (kết trạng thái); yêu cầu co-partition.
• processing.guarantee=exactly_once_v2 cho exactly-once trong phạm vi Kafka nhờ giao dịch nguyên tử.
• ksqlDB viết cùng logic bằng SQL (CREATE STREAM/TABLE, SELECT ... EMIT CHANGES) — đây là SQL của ksqlDB, không phải SQL cơ sở dữ liệu quan hệ.

Tự kiểm tra

1. Giải thích vì sao KStream và KTable được gọi là "nhị nguyên", và mối liên hệ của KTable với một topic có cleanup policy = compact?
2. Một toán tử stateful như count lưu trạng thái ở đâu, và làm thế nào trạng thái đó được phục hồi khi một instance bị crash?
3. Phân biệt cửa sổ tumbling và hopping; khi nào dữ liệu một bản ghi có thể thuộc nhiều cửa sổ?
4. grace period giải quyết vấn đề gì, và đánh đổi khi đặt grace dài so với grace ngắn là gì?
5. Vì sao stream–stream join bắt buộc phải có cửa sổ, còn stream–table join thì không?
6. exactly_once_v2 đảm bảo điều gì và trong phạm vi nào; nó có tự động đảm bảo idempotency khi bạn ghi ra một cơ sở dữ liệu bên ngoài không?

Đọc tiếp

Kafka 7 — Vận hành Production