Flink 2 — Mô hình Dataflow & DataStream API

Chương trình Flink là một đồ thị dataflow

Điểm khác biệt cốt lõi khi lập trình Flink so với viết code tuần tự thông thường: bạn không ra lệnh "làm A xong rồi làm B". Thay vào đó, bạn mô tả một đồ thị luồng dữ liệu (dataflow graph) — dữ liệu chảy qua đâu, bị biến đổi thế nào — rồi Flink runtime nhận đồ thị đó, tối ưu và thực thi phân tán trên cluster.

Mọi chương trình Flink đều là chuỗi ba loại thành phần:

• Source: nơi dữ liệu đi vào (Kafka topic, file, socket, generator...).
• Transformation: các phép biến đổi trên luồng (map, filter, keyBy, window, process...).
• Sink: nơi kết quả đi ra (Kafka, database, file, console...).

Nối các thành phần này lại ta được một streaming dataflow — một đồ thị có hướng, dữ liệu chảy từ source qua các transformation tới sink, không có điểm dừng vì luồng là vô tận (unbounded). Ở bài 1 chúng ta đã thấy vì sao Flink coi batch chỉ là trường hợp đặc biệt (bounded) của streaming; ở bài này ta đi sâu vào cách biểu diễn và thực thi đồ thị đó.

Logical graph vs Physical (Execution) graph

Khi bạn viết code, bạn tạo ra logical dataflow graph (còn gọi StreamGraph rồi rút gọn thành JobGraph): mỗi operator là một node logic, mỗi phép biến đổi là một cạnh. Đây là "ý định" của chương trình, độc lập với việc chạy trên bao nhiêu máy.

Khi submit job, JobManager biên dịch logical graph thành physical graph (ExecutionGraph): mỗi operator logic được nhân bản thành nhiều subtask theo độ song song (parallelism), rồi phân bổ vào các task slot trên các TaskManager. Đây mới là thứ thực sự chạy.

Ví dụ: một operator map với parallelism = 4 trong logical graph sẽ trở thành 4 subtask chạy song song trong execution graph, mỗi subtask xử lý một phần luồng.

Chú ý cạnh giữa map và process: vì có keyBy ở giữa, dữ liệu phải phân phối lại theo hash của khoá (mỗi bản ghi của account X luôn về đúng một subtask process), nên mỗi source-subtask nối tới mọi process-subtask. Ta sẽ nói kỹ về partitioning ở dưới.

DataStream API: Source, Transformation, Sink

DataStream<T> là kiểu dữ liệu trung tâm — đại diện cho một luồng vô tận các phần tử kiểu T. Bạn xây pipeline bằng cách gọi các phương thức biến đổi trên DataStream, mỗi lần trả về một DataStream mới.

Source

Source hiện đại dùng interface Source thống nhất (fromSource(...)). Vài nguồn thường gặp:

• Kafka: đọc từ topic — nguồn phổ biến nhất trong ngân hàng (xem Kafka kiến trúc).
• File: đọc file/thư mục (bounded hoặc theo dõi thư mục — streaming).
• Socket: socketTextStream(host, port) — chỉ dùng để thử nghiệm, không dùng production.
• Collection / generator: tạo dữ liệu tổng hợp để test.

// Minh hoạ (Java) — đọc giao dịch từ Kafka
KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("txn-events")
    .setGroupId("flink-fraud")
    .setStartingOffsets(OffsetsInitializer.latest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

DataStream<String> raw = env.fromSource(
    source, WatermarkStrategy.noWatermarks(), "kafka-source");

Transformation

Là phần thân của pipeline — biến DataStream này thành DataStream khác. Chi tiết ở mục sau.

Sink

Nơi ghi kết quả ra ngoài: Kafka sink, JDBC sink (ghi vào Postgres/Oracle), file sink, hoặc .print() khi debug. Sink là điểm mà Flink phối hợp với cơ chế checkpoint để đảm bảo ngữ nghĩa exactly-once end-to-end (chi tiết ở bài 6).

stream.sinkTo(KafkaSink.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(/* ... */)
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .build());

Các phép biến đổi cơ bản

map / flatMap / filter

Các phép element-wise — xử lý từng bản ghi độc lập, không cần state, không xáo trộn dữ liệu (partitioning kiểu forward).

• map: 1 vào → 1 ra. VD: parse JSON thành object giao dịch.
• flatMap: 1 vào → 0..N ra. VD: tách một message chứa nhiều dòng.
• filter: giữ lại phần tử thoả điều kiện. VD: chỉ giữ giao dịch amount > 0.

# Minh hoạ (PyFlink)
txns = raw.map(lambda s: parse_txn(s), output_type=Types.PICKLED_BYTE_ARRAY())
big   = txns.filter(lambda t: t.amount >= 100_000_000)  # >= 100 triệu VND

keyBy — phân vùng theo khoá

keyBy là phép then chốt nhất vì nó là nền cho state và window. keyBy(k) chia luồng thành các keyed stream logic: mọi bản ghi có cùng khoá k được định tuyến (hash-partitioned) tới cùng một subtask. Nhờ vậy các phép sau đó (reduce, aggregate, window, process với state) có thể giữ trạng thái riêng theo từng khoá một cách an toàn.

// gom giao dịch theo tài khoản — mọi giao dịch của 1 account về cùng 1 subtask
KeyedStream<Txn, String> byAccount = txns.keyBy(t -> t.accountId);

Trong ngân hàng, khoá thường là account_id, customer_id, card_id hoặc terminal_id. Chọn khoá quyết định cả tính đúng đắn (state phải gom đúng thực thể) lẫn hiệu năng (khoá phân bố lệch → data skew, một subtask quá tải).

reduce / aggregate

Trên keyed stream, reduce gộp dần các phần tử cùng khoá thành một kết quả tích luỹ; aggregate linh hoạt hơn với accumulator kiểu riêng (VD tính trung bình, đếm phân biệt). Đây là các phép stateful: Flink giữ giá trị tích luỹ theo từng khoá.

# tổng tiền luỹ kế theo tài khoản (minh hoạ)
running_total = txns.key_by(lambda t: t.account_id) \
    .reduce(lambda a, b: Txn(a.account_id, a.amount + b.amount))

union / connect / coMap

• union: gộp nhiều luồng cùng kiểu thành một. VD: gộp giao dịch thẻ + giao dịch chuyển khoản (nếu cùng schema).
• connect: nối hai luồng khác kiểu thành ConnectedStream, cho phép hai luồng chia sẻ state. Rất hữu ích khi một luồng là dữ liệu chính, luồng kia là control/rules (VD: luồng cấu hình ngưỡng cảnh báo cập nhật động).
• coMap / CoProcessFunction: xử lý hai luồng đã connect với hai hàm riêng cho mỗi luồng.

// stream giao dịch connect với stream ngưỡng cảnh báo động
ConnectedStreams<Txn, Rule> conn = txns.connect(rulesStream);
DataStream<Alert> alerts = conn
    .keyBy(t -> t.accountId, r -> r.accountId)
    .process(new FraudRuleFunction());  // giữ rule trong state, áp lên txn

process — hàm cấp thấp

process (ProcessFunction / KeyedProcessFunction) là API linh hoạt nhất: truy cập trực tiếp state, timestamp, watermark, và timer (đăng ký để kích hoạt logic tại một mốc thời gian). Đây là nơi bạn viết logic phát hiện gian lận phức tạp mà map/reduce không đủ sức. Timer + event-time sẽ nối sang bài 3.

Parallelism, task slot & subtask

Parallelism của một operator là số bản (subtask) chạy song song của nó. Đặt ở nhiều cấp: mặc định cluster → mặc định job (env.setParallelism(N)) → từng operator (.setParallelism(N) sau phép biến đổi).

• Subtask: một bản thực thi của operator, xử lý một slice của luồng. Operator parallelism = 4 → 4 subtask.
• Task slot: đơn vị tài nguyên trên một TaskManager. Mỗi TaskManager có cố định số slot; mỗi slot chạy một "pipeline" các subtask. Tổng số slot của cluster giới hạn parallelism tối đa của job.
• Nhờ slot sharing (mặc định), các subtask của các operator khác nhau trong cùng job có thể dùng chung một slot, giúp tận dụng tài nguyên và giảm số slot cần thiết.

Ví dụ: cluster có 3 TaskManager, mỗi TM 4 slot → 12 slot. Job đặt parallelism = 8 sẽ cần 8 slot (còn dư 4). Nếu đặt parallelism = 16 nhưng chỉ có 12 slot → job không lập lịch được.

Operator chaining

Để giảm chi phí, Flink gộp (chain) nhiều operator liên tiếp vào cùng một task khi có thể — chẳng hạn source → map → filter nếu chúng cùng parallelism và nối theo kiểu forward. Lợi ích:

• Không phải serialize/deserialize dữ liệu giữa các operator được chain.
• Không chuyển qua buffer/mạng — các operator chạy trong cùng thread, gọi hàm trực tiếp.
• Giảm độ trễ và tăng thông lượng đáng kể.

Chaining bị cắt khi giữa hai operator có shuffle (VD sau keyBy phải re-partition theo hash) hoặc khi parallelism thay đổi. Bạn có thể can thiệp thủ công: .disableChaining(), .startNewChain(), hoặc env.disableOperatorChaining() (thường để debug/đo lường từng operator).

Partitioning: dữ liệu đi giữa các operator thế nào

Khi dữ liệu chuyển từ subtask của operator này sang operator sau, Flink dùng một chiến lược phân vùng để quyết định bản ghi nào về subtask nào:

hash, rebalance, broadcast đều là redistributing — dữ liệu qua buffer + mạng, chi phí cao hơn forward. Vì thế keyBy "đắt" hơn map: nó luôn cắt chaining và gây shuffle. Nguyên tắc thực chiến: keyBy một lần, tái sử dụng keyed stream thay vì keyBy lặp đi lặp lại theo cùng khoá.

Data exchange & record đi kèm timestamp

Giữa các operator (khi không chain), dữ liệu được đóng gói vào network buffer và truyền qua kênh có kiểm soát backpressure — nếu operator sau chậm, tín hiệu áp ngược lan về source, làm chậm cả pipeline một cách an toàn thay vì tràn bộ nhớ.

Điều quan trọng cho các bài sau: mỗi record (bản ghi) chảy trong pipeline có thể mang theo một timestamp (thường là event time — thời điểm sự kiện thực sự xảy ra, lấy từ chính dữ liệu, VD created_at của giao dịch). Cùng với đó, Flink chèn các watermark vào luồng để đánh dấu tiến độ thời gian. Cặp record+timestamp và watermark chính là nền tảng cho windowing theo event-time — chủ đề trọng tâm của bài 3. State bền vững (nơi keyed stream lưu trạng thái) được đào sâu ở bài 4.

Use case thực tế

Bối cảnh (minh hoạ tại NCB): phát hiện giao dịch bất thường gần thời gian thực trên luồng ~2.000 giao dịch/giây từ core banking, đẩy qua Kafka topic txn-events.

Pipeline dataflow thiết kế:

1. Source (parallelism 6): đọc txn-events từ Kafka, gán event-time = created_at của giao dịch.
2. map + filter (chained với source): parse JSON, loại giao dịch giá trị 0 và giao dịch nội bộ hệ thống. Vì cùng parallelism và forward → được chain, không tốn shuffle.
3. keyBy(account_id): chuyển sang keyed stream — mọi giao dịch của một tài khoản về cùng subtask. Đây là điểm shuffle (hash-partition), cắt chaining.
4. KeyedProcessFunction (parallelism 6): giữ state theo tài khoản (tổng tiền 5 phút gần nhất, số lần rút, quốc gia gần nhất), đăng ký timer phát cảnh báo khi vượt ngưỡng. Một luồng rules được connect để cập nhật ngưỡng động cho từng phân khúc khách hàng mà không phải restart job.
5. Sink: ghi cảnh báo vào Kafka topic fraud-alerts với EXACTLY_ONCE, đồng thời một JDBC sink ghi log vào Postgres cho đội vận hành.

Con số minh hoạ: cluster 3 TaskManager × 4 slot = 12 slot; job parallelism 6 dùng 6 slot, còn dư để chạy song song một job đối soát. Nhờ chaining source+map+filter, mỗi bản ghi chỉ serialize một lần tại bước keyBy thay vì ba lần → giảm ~30% chi phí CPU serialize so với khi tắt chaining (đo lường minh hoạ). Với khoá account_id, cần theo dõi skew: nếu một tài khoản gom quỹ (VD tài khoản trung gian) chiếm quá nhiều lưu lượng, subtask tương ứng sẽ nghẽn — giải pháp là tách khoá phụ hoặc rebalance ở bước tiền xử lý.

Ghi nhớ

• Chương trình Flink = đồ thị dataflow source → transformation → sink; bạn mô tả đồ thị, runtime thực thi nó.
• Logical graph (operator) được JobManager biên dịch thành execution graph (nhiều subtask theo parallelism) rồi phân bổ vào task slot.
• DataStream<T> là kiểu trung tâm; source hiện đại dùng fromSource (Kafka, file, socket, generator).
• keyBy là phép then chốt: hash-partition theo khoá, mọi bản ghi cùng khoá về cùng subtask — nền tảng cho state và window. keyBy luôn gây shuffle và cắt chaining.
• map/flatMap/filter là stateless & forward; reduce/aggregate là stateful trên keyed stream; connect/coProcess nối 2 luồng khác kiểu chia sẻ state; process cho toàn quyền với state + timer.
• Parallelism = số subtask; task slot = tài nguyên trên TaskManager; slot sharing giúp nhiều operator dùng chung slot.
• Operator chaining gộp operator forward cùng parallelism vào một task để tránh serialize/mạng — giảm độ trễ, tăng thông lượng.
• Partitioning: forward (rẻ, chain được) vs hash/rebalance/broadcast/rescale (redistributing, tốn shuffle mạng).
• Mỗi record có thể mang timestamp (event-time); cùng watermark là nền cho windowing ở bài 3.