Kafka Streams là một thư viện xử lý luồng (stream processing library)

Kafka Streams là một thư viện xử lý luồng (stream processing library) bằng Java/Scala do chính đội ngũ phát triển Apache Kafka tạo ra. Điểm khác biệt cốt lõi nhất của Kafka Streams so với các nền tảng xử lý dữ liệu lớn như Apache Flink hay Apache Spark là nó không cần một cụm máy chủ tính toán riêng biệt (compute cluster). Nó chỉ là một thư viện bạn nhúng thẳng vào ứng dụng microservice của mình, chạy ở bất cứ đâu từ máy cá nhân, Docker container cho đến Kubernetes.

Dưới đây là cẩm nang toàn diện từ kiến trúc cốt lõi, các khái niệm nâng cao cho đến ví dụ thực chiến và so sánh thực tế về Kafka Streams.

1. Kiến trúc cốt lõi: Processor Topology

Trái tim của mọi ứng dụng Kafka Streams là Processor Topology (Cấu trúc liên kết xử lý). Trong khoa học máy tính, nó được biểu diễn dưới dạng một Đồ thị có hướng không chu trình (DAG - Directed Acyclic Graph), bao gồm các điểm nút (Nodes) nối với nhau bằng các luồng dữ liệu (Streams):

[Kafka Input Topic] 
        │
        ▼
┌───────────────────┐
│ Source Processor  │  (Đọc dữ liệu từ Kafka)
└─────────┬─────────┘
          │
          ▼
┌───────────────────┐      ┌─────────────────┐
│ Stream Processor  │◄────►│   State Store   │ (RocksDB / In-Memory)
│ (Filter/Map/Join) │      └────────┬────────┘
└─────────┬─────────┘               │ (Sao lưu tự động)
          │                         ▼
          ▼                [Changelog Topic] (Trong Kafka)
┌───────────────────┐
│  Sink Processor   │  (Ghi kết quả ra Kafka)
└─────────┬─────────┘
          │
          ▼
[Kafka Output Topic]

• Source Processor: Điểm nút đầu vào, không có nút cha. Nó tiêu thụ (consume) các bản ghi trực tiếp từ một hoặc nhiều Kafka Topic và chuyển tiếp cho các nút phía sau.

• Stream Processor: Điểm nút trung gian thực hiện logic nghiệp vụ như lọc (filter), biến đổi (map), gom nhóm (groupBy), hoặc tổng hợp (aggregate).

• Sink Processor: Điểm nút cuối cùng, không có nút con. Nó nhận dữ liệu đã xử lý xong và đẩy (produce) ngược trở lại vào một Kafka Topic đích.

Để xây dựng Topology này, Kafka Streams cung cấp 2 giao diện lập trình (API):

1. Kafka Streams DSL (Domain Specific Language): API bậc cao, cung cấp sẵn các hàm lập trình hàm (functional programming) như map, filter, join, aggregate. 90% các bài toán thực tế chỉ cần dùng DSL.

2. Processor API: API bậc thấp (low-level), cho phép lập trình viên tự định nghĩa từng điểm nút, tự quản lý State Store và lập lịch chạy tác vụ định kỳ (Punctuators). Dùng khi bài toán có logic quá đặc thù mà DSL không hỗ trợ sẵn.

2. Sự đối ngẫu giữa Luồng và Bảng (Stream-Table Duality)

Để xử lý dữ liệu thời gian thực hiệu quả, bạn bắt buộc phải hiểu mối quan hệ hai chiều giữa Luồng sự kiện (Stream) và Bảng trạng thái (Table). Kafka Streams cụ thể hóa khái niệm này qua 3 đối tượng:

Ví dụ minh họa sự khác biệt giữa KStream và KTable: Giả sử có 3 bản ghi lần lượt đi vào hệ thống: ("userA", 10), ("userB", 20), ("userA", 50).

• Nếu đọc dưới dạng KStream và tính tổng: Hệ thống hiểu là 3 giao dịch rời rạc $\rightarrow$ userA có tổng là $10 + 50 = 60$, userB là $20$.

• Nếu đọc dưới dạng KTable: Hệ thống hiểu là cập nhật trạng thái mới nhất $\rightarrow$ Bản ghi ("userA", 50) sẽ đè lên ("userA", 10). Trạng thái hiện tại: userA = 50, userB = 20.

3. Các phép toán xử lý dữ liệu (Operations)

Các phép toán trong Kafka Streams được chia làm 2 nhóm lớn dựa trên việc chúng có cần lưu nhớ dữ liệu quá khứ hay không:

3.1. Stateless Operations (Xử lý không trạng thái)

Mỗi bản ghi đi qua được xử lý hoàn toàn độc lập, không cần biết trước đó đã có bản ghi nào đi qua hay chưa. Ưu điểm là tốc độ cực nhanh và không tốn bộ nhớ lưu trữ cục bộ:

• filter / filterNot: Giữ lại hoặc loại bỏ các bản ghi thỏa mãn điều kiện.

• map / mapValues: Biến đổi một bản ghi thành một bản ghi mới. (Lưu ý: Luôn ưu tiên dùng mapValues thay vì map nếu không cần đổi Key, vì việc thay đổi Key sẽ kích hoạt quá trình Repartitioning - gửi dữ liệu qua mạng về lại Kafka để chia lại phân vùng, gây tốn băng thông).

• flatMap / flatMapValues: Nhận 1 bản ghi đầu vào và tách ra thành 0, 1 hoặc nhiều bản ghi đầu ra (ví dụ: tách 1 câu văn thành danh sách các từ).

• branch / split: Rẽ nhánh một luồng dữ liệu thành nhiều luồng con dựa trên các điều kiện khác nhau.

3.2. Stateful Operations (Xử lý có trạng thái)

Để tính tổng, đếm, hoặc kết hợp (Join) hai luồng dữ liệu, ứng dụng bắt buộc phải "nhớ" các sự kiện đã xảy ra trước đó:

• Aggregation (count, reduce, aggregate): Gom nhóm dữ liệu theo Key (groupByKey hoặc groupBy) và tính toán tích lũy. Kết quả của một phép Aggregation luôn trả về một KTable.

• Joining (Kết hợp luồng):

  • KStream - KStream Join: Luôn là Windowed Join (Join theo cửa sổ thời gian) vì luồng là vô tận. Ví dụ: Khớp sự kiện "Đặt hàng" và sự kiện "Thanh toán" xảy ra trong vòng 15 phút của cùng một đơn hàng.

  • KStream - KTable Join: Thường dùng để làm giàu dữ liệu (Data Enrichment). Ví dụ: Luồng sự kiện "Mua hàng" (chỉ chứa user_id) được Join với bảng KTable "Thông tin khách hàng" để lấy thêm tên, địa chỉ, hạng thẻ VIP.

  • KTable - KTable Join: Hoạt động giống hệt phép Join giữa 2 bảng trong cơ sở dữ liệu quan hệ (RDBMS).

4. Quản lý Trạng thái (State Store) và Khả năng Chịu lỗi

Khi thực hiện các phép toán Stateful (như đếm số đơn hàng của từng khách), Kafka Streams lưu trữ trạng thái đó ở đâu để nếu sập nguồn thì không bị mất?

1. State Store cục bộ (RocksDB): Mặc định, Kafka Streams nhúng sẵn RocksDB (một cơ sở dữ liệu Key-Value hiệu năng cao viết bằng C++ của Facebook) ngay trên ổ cứng của máy đang chạy ứng dụng (hoặc có thể cấu hình lưu trên RAM với In-Memory Store). Việc tra cứu diễn ra ngay tại máy cục bộ nên tốc độ cực kỳ nhanh, không tốn độ trễ mạng như khi gọi ra Redis hay Database bên ngoài.

2. Cơ chế tự sao lưu qua Changelog Topic: Mọi thay đổi ghi vào RocksDB cục bộ sẽ đồng thời được Kafka Streams âm thầm gửi vào một Kafka Topic đặc biệt gọi là Changelog Topic (được bật tính năng Log Compaction - chỉ giữ lại giá trị mới nhất của mỗi Key).

3. Tự phục hồi (Fault Tolerance) và Standby Replicas:

   • Nếu Instance A đang chạy mà bị chết (Crash), Kafka Streams sẽ tự động chuyển công việc của Instance A sang cho Instance B đang sống. Instance B sẽ đọc lại dữ liệu từ Changelog Topic để tái tạo lại RocksDB y hệt như lúc Instance A chưa chết.

   • Để tránh thời gian chờ tải lại dữ liệu (rebuilding state) khi gặp sự cố, bạn có thể cấu hình tham số num.standby.replicas = 1. Khi đó, Kafka Streams sẽ duy trì một bản sao trạng thái "nóng" (hot standby) ở một máy khác, sẵn sàng tiếp quản công việc ngay lập tức trong vài mili-giây.

4. Interactive Queries (Truy vấn trực tiếp): Thay vì phải đẩy kết quả ra một Database bên ngoài để làm báo cáo, Kafka Streams cho phép các ứng dụng bên ngoài (như API Server) gọi trực tiếp vào State Store cục bộ của nó để đọc dữ liệu tính toán tức thời.

5. Khái niệm Thời gian (Time) và Cửa sổ (Windowing)

Trong xử lý luồng, một sự kiện tạo ra lúc 10:00:00 trên điện thoại người dùng (do mất mạng 3G) có thể đến 10:05:00 mới tới được máy chủ Kafka. Vì vậy, Kafka Streams phân biệt 3 khái niệm thời gian:

1. Event Time (Thời gian sự kiện): Thời điểm sự kiện thực sự xảy ra tại nguồn phát sinh (được gắn vào metadata của bản ghi hoặc nằm trong payload dữ liệu). Đây là loại thời gian quan trọng nhất trong phân tích nghiệp vụ.

2. Ingestion Time (Thời gian tiếp nhận): Thời điểm bản ghi được Kafka Broker lưu vào phân vùng (Partition).

3. Processing Time (Thời gian xử lý): Thời điểm ứng dụng Kafka Streams thực sự đọc và xử lý bản ghi đó (giờ đồng hồ của máy chạy ứng dụng).

Các loại Cửa sổ thời gian (Windowing)

Khi cần gom nhóm dữ liệu theo khung giờ (ví dụ: "Đếm số lượt truy cập mỗi 5 phút"), Kafka Streams cung cấp 4 loại cửa sổ:

1. Tumbling Window (Cố định, không gối đầu, nối tiếp nhau):
|--- 5 phút ---|--- 5 phút ---|--- 5 phút ---|

2. Hopping Window (Cố định 5 phút, trượt mỗi 1 phút -> có trùng lặp):
|--- 5 phút ---|
   |--- 5 phút ---|
      |--- 5 phút ---|

3. Sliding Window (Kích hoạt động dựa trên khoảng cách thời gian giữa 2 bản ghi):
   [Sự kiện A] <--- <= 5 phút ---> [Sự kiện B]

4. Session Window (Dựa trên hành vi hoạt động, ngắt khi không có tương tác - Inactivity Gap):
|-- Phiên 1 --|  (Nghỉ > 30p)  |----- Phiên 2 -----|

• Grace Period (Thời gian ân hạn): Dùng để xử lý các bản ghi đến muộn (Late-arriving records). Ví dụ: Cửa sổ từ 10:00 đến 10:05 có cấu hình grace(Duration.ofMinutes(2)). Nếu một sự kiện có Event Time là 10:04 nhưng do nghẽn mạng đến tận 10:06:30 mới tới nơi, hệ thống vẫn chấp nhận cộng dồn nó vào cửa sổ 10:00 - 10:05. Nếu nó tới lúc 10:07:01 (quá hạn 2 phút), bản ghi đó sẽ bị bỏ qua.

6. Cơ chế Mở rộng quy mô (Scaling): Partitions, Tasks và Threads

Cách Kafka Streams chia nhỏ công việc để chạy song song hoàn toàn dựa trên kiến trúc phân vùng (Partition) của Kafka:

• Stream Task (Đơn vị công việc nhỏ nhất): Khi ứng dụng khởi chạy, Kafka Streams kiểm tra xem các Topic đầu vào có bao nhiêu Partitions, từ đó nó tạo ra bấy nhiêu Stream Tasks.

  • Ví dụ: Topic đầu vào có 6 Partitions $\rightarrow$ Kafka Streams tạo ra cố định 6 Tasks (Task 0 xử lý Partition 0, ..., Task 5 xử lý Partition 5). Mỗi Task có một State Store (RocksDB) riêng biệt, không dùng chung để tránh tranh chấp khóa (Lock contention).

• Stream Thread (Luồng thực thi): Bạn có thể cấu hình số luồng chạy trong 1 instance ứng dụng qua tham số num.stream.threads (hoặc bật thêm các instance mới trên các máy chủ khác có cùng application.id).

  • Nếu chạy 1 Instance (1 Thread): Luồng duy nhất này gánh cả 6 Tasks.

  • Nếu bật thêm Instance thứ 2 (cùng application.id): Kafka tự động Rebalance (cân bằng tải), chia cho mỗi Instance chạy 3 Tasks.

  • Nếu bật tối đa 6 Instances: Mỗi Instance xử lý đúng 1 Task (Tốc độ tối đa).

  • Nếu bật Instance thứ 7: Instance thứ 7 sẽ ở trạng thái rơi vào trạng thái nghỉ (Idle) vì chỉ có tối đa 6 Tasks (tương ứng 6 Partitions). Đây là giới hạn vật lý bạn cần nhớ khi thiết kế số lượng Partition ban đầu.

7. Cam kết xử lý chính xác một lần (Exactly-Once Semantics - EOS)

Trong hệ thống phân tán, việc mạng chập chờn hoặc ứng dụng khởi động lại rất dễ dẫn đến việc một bản ghi bị tính toán 2 lần (ví dụ: trừ tiền khách hàng 2 lần). Kafka Streams hỗ trợ 2 chế độ đảm bảo toàn vẹn dữ liệu thông qua cấu hình processing.guarantees:

1. at_least_once (Mặc định - Ít nhất một lần): Đảm bảo không bao giờ mất dữ liệu, nhưng nếu ứng dụng sập đột ngột giữa chừng trước khi kịp lưu vị trí đọc (commit offset), khi bật lại nó có thể xử lý lặp lại một vài bản ghi.

2. exactly_once_v2 (Chính xác một lần): Khi bật chỉ với 1 dòng cấu hình duy nhất, Kafka Streams sẽ sử dụng cơ chế Transactional API ở tầng dưới để gom 3 hành động: (1) Đọc và lưu vị trí Offset đầu vào + (2) Cập nhật trạng thái vào Changelog Topic + (3) Ghi kết quả ra Output Topic thành một Giao dịch nguyên tử (Atomic Transaction). Nếu bất kỳ bước nào lỗi, toàn bộ giao dịch được Rollback như chưa từng xảy ra, đảm bảo mỗi sự kiện chỉ tác động đến kết quả cuối cùng đúng 1 lần duy nhất.

8. Ví dụ Thực chiến bằng Code Java

Dưới đây là một ứng dụng Kafka Streams hoàn chỉnh viết bằng Java sử dụng thư viện chuẩn: Đọc một luồng các câu văn từ topic input-text, tách từng từ, đếm số lần xuất hiện của mỗi từ theo thời gian thực và xuất kết quả ra topic word-count-output.

Bước 1: Khai báo thư viện trong pom.xml (Maven)

XML

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>3.7.0</version>
</dependency>

Bước 2: Code xử lý luồng (WordCountApp.java)

Java

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;

import java.util.Arrays;
import java.util.Properties;

public class WordCountApp {
    public static void main(String[] args) {
        // 1. Cấu hình thông số cho ứng dụng Kafka Streams
        Properties props = new Properties();
        // application.id đóng vai trò như group.id của Consumer và định danh các State Store
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-microservice-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        
        // Khai báo bộ tuần tự hóa/giải tuần tự hóa (Serde) mặc định cho Key và Value
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        
        // (Tùy chọn) Bật chế độ xử lý chính xác 1 lần (Exactly-Once)
        props.put(StreamsConfig.processingGuaranteeConfig(), StreamsConfig.EXACTLY_ONCE_V2);

        // 2. Xây dựng Processor Topology bằng DSL
        StreamsBuilder builder = new StreamsBuilder();

        // Đọc luồng dữ liệu từ topic "input-text"
        KStream<String, String> textLines = builder.stream("input-text");

        // Xử lý dữ liệu: Tách câu thành từ -> Gán từ làm Key -> Đếm số lượng
        KTable<String, Long> wordCounts = textLines
            // Chuyển thành chữ thường và tách câu thành mảng các từ
            .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
            // Lọc bỏ các từ rỗng
            .filter((key, word) -> !word.isEmpty())
            // Gom nhóm theo chính từ đó (chuyển Value thành Key mới để đếm)
            .groupBy((key, word) -> word)
            // Đếm và lưu trạng thái vào State Store (RocksDB) có tên "counts-store"
            .count(Materialized.as("counts-store"));

        // Chuyển KTable kết quả ngược lại thành KStream và ghi ra topic "word-count-output"
        wordCounts.toStream().to("word-count-output", Produced.with(Serdes.String(), Serdes.Long()));

        // 3. Khởi tạo và chạy ứng dụng
        KafkaStreams streams = new KafkaStreams(builder.build(), props);

        // Đăng ký Shutdown Hook để đóng kết nối và dọn dẹp RocksDB an toàn khi tắt app
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

        // Bắt đầu chạy luồng xử lý
        streams.start();
    }
}

Kiểm thử nhanh mà không cần bật Kafka (TopologyTestDriver): Một điểm cộng rất lớn cho các lập trình viên là Kafka Streams cung cấp sẵn bộ công cụ kafka-streams-test-utils. Bạn có thể viết Unit Test giả lập đẩy dữ liệu vào builder.build() và kiểm tra đầu ra ngay trong bộ nhớ chỉ mất vài mili-giây mà không cần cài đặt hay chạy Kafka Broker hay Docker.

9. So sánh: Khi nào dùng Kafka Streams, Flink hay Spark?

Không có công cụ nào là hoàn hảo cho mọi bài toán. Bảng so sánh dưới đây sẽ giúp bạn chọn đúng công cụ cho hệ thống của mình:

Lời khuyên lựa chọn thực tế:

• Chọn Kafka Streams khi: Bạn đang xây dựng các Microservices bằng Java/Scala/Spring Boot, dữ liệu đầu vào và đầu ra đều đã nằm trên Kafka, và bạn muốn xử lý các nghiệp vụ thời gian thực (Event-Driven Architecture) mà không muốn gánh thêm chi phí vận hành một cụm server Big Data cồng kềnh.

• Chọn Apache Flink khi: Bạn làm việc trong đội ngũ Data Engineering chuyên sâu, cần nối dữ liệu từ nhiều nguồn ngoài Kafka (như đổ thẳng vào Data Lakehouse), lượng trạng thái (State) lớn tới hàng chục Terabytes, hoặc cần viết quy trình xử lý luồng bằng SQL / Python.

• Chọn Apache Spark khi: Hệ thống của bạn vốn đã dùng Spark để chạy các tác vụ tổng hợp dữ liệu theo lô (Batch Processing), làm Machine Learning, và bạn muốn tận dụng lại chung một hạ tầng đó để xử lý luồng gần thời gian thực (Near real-time).