Flink 8 — Use case ngân hàng & CEP

Ghép các mảnh lại: Flink làm gì trong một ngân hàng

Bảy bài trước dựng nền: tổng quan, dataflow, event-time & window, state, Flink SQL, fault tolerance và vận hành. Bài này tổng kết ứng dụng: gom mọi khái niệm đó lại và trả lời câu hỏi thực tế nhất — một ngân hàng như NCB dùng Flink để làm gì, và bài toán nào thì Flink thắng, bài toán nào công cụ khác gọn hơn.

Điểm chung của mọi use case dưới đây: dữ liệu có giá trị lớn nhất ngay lúc nó xảy ra. Một giao dịch gian lận chỉ đáng chặn trước khi tiền rời tài khoản; một chuỗi đăng nhập bất thường chỉ đáng cảnh báo khi đang diễn ra. Batch cuối ngày luôn trễ. Flink lấp đúng khoảng đó: xử lý stateful, độ trễ mili giây, phát hiện được cả mẫu chuỗi sự kiện chứ không chỉ tổng hợp đơn lẻ.

Use case 1 — Phát hiện gian lận real-time

Đây là use case đinh của Flink trong ngân hàng. Yêu cầu: khi một giao dịch (thẻ, chuyển khoản, rút ATM) đi vào, hệ thống phải chấm điểm rủi ro và quyết định chặn/cho qua trong vài chục mili giây — trước khi giao dịch hoàn tất. Nếu chậm, tiền đã đi.

Vì sao Flink hợp:

• Stateful theo tài khoản. Chấm điểm cần context lịch sử: tài khoản này bình thường tiêu bao nhiêu, ở đâu, giờ nào. Flink giữ keyed state theo account_id (xem Flink 4) — đếm giao dịch 5 phút, tổng tiền 1 giờ, merchant gần đây — cập nhật tức thì mỗi giao dịch, không join lại DB mỗi lần.
• Feature real-time. Các đặc trưng như "số giao dịch/phút", "khoảng cách địa lý giữa hai giao dịch liên tiếp", "lệch so với chi tiêu trung bình" được tính ngay trên luồng, làm đầu vào cho mô hình chấm điểm (rule-based hoặc ML).
• Độ trễ mili giây. Pipeline Kafka → Flink → quyết định có độ trễ đầu-cuối rất thấp, đủ để chặn đồng bộ.

Luồng điển hình: giao dịch vào Kafka → Flink làm giàu (thêm hồ sơ khách, hạng thẻ), tính feature real-time từ state, chạy luật + điểm ML → nếu điểm vượt ngưỡng thì phát cảnh báo/lệnh chặn ra một topic Kafka mà hệ thống core banking đang chờ. Chi tiết nghiệp vụ chống gian lận xem AML — giám sát giao dịch.

Use case 2 — Giám sát giao dịch & AML real-time

Chống rửa tiền (AML — Anti-Money Laundering) và giám sát tuân thủ vốn nặng về batch cuối ngày, nhưng nhiều luật có bản real-time hiệu quả hơn nhiều:

• Velocity check (kiểm tra tốc độ). Đếm số giao dịch hoặc tổng tiền của một tài khoản trong cửa sổ trượt ngắn; vượt ngưỡng thì cảnh báo (ví dụ >20 giao dịch/10 phút, hoặc >500 triệu chuyển ra/1 giờ). Đây là sliding/keyed window + state đếm — đúng sở trường Flink.
• Structuring (smurfing — chia nhỏ để né ngưỡng). Kẻ rửa tiền chia một khoản lớn thành nhiều giao dịch nhỏ dưới ngưỡng báo cáo. Real-time, Flink phát hiện "nhiều giao dịch nhỏ cùng chiều trong thời gian ngắn cộng lại vượt ngưỡng" — luật khó nếu chỉ nhìn từng giao dịch rời.
• Giám sát dòng tiền vòng (layering). Tiền chạy qua chuỗi tài khoản A→B→C→A trong thời gian ngắn — cần nhìn chuỗi sự kiện, dẫn thẳng sang CEP dưới đây.

Điểm mấu chốt: velocity và structuring không phải "một truy vấn tổng hợp" thuần — chúng là quyết định theo ngưỡng trên state trượt, và nhiều luật AML mạnh nhất lại là mẫu chuỗi, thứ mà SQL GROUP BY khó diễn đạt. Đó là lý do phải nói về CEP.

Use case 3 — Complex Event Processing (CEP)

Complex Event Processing (CEP — xử lý sự kiện phức hợp) là bài toán phát hiện một MẪU CHUỖI sự kiện trên luồng: không hỏi "có bao nhiêu giao dịch" mà hỏi "có xảy ra chuỗi A rồi B rồi C trong vòng T không". Vài ví dụ ngân hàng kinh điển:

• 3 lần đăng nhập sai rồi một lần rút tiền lớn trong vòng 10 phút → dấu hiệu chiếm đoạt tài khoản (account takeover).
• Nhiều giao dịch nhỏ liên tiếp trong thời gian rất ngắn (thử thẻ — card testing) → gian lận thẻ.
• Hai giao dịch cách nhau bất khả thi về địa lý (Hà Nội rồi 3 phút sau ở nước ngoài) → thẻ bị sao chép.
• Đăng nhập lạ → đổi thông tin liên hệ → yêu cầu chuyển tiền trong vài phút → lừa đảo tiếp quản tài khoản.

Flink có hai cách làm CEP:

Flink CEP library (DataStream)

Thư viện flink-cep cho phép định nghĩa Pattern bằng API trên DataStream. Các khối cơ bản:

• begin("A") — bắt đầu mẫu bằng một sự kiện thoả điều kiện A (ví dụ event.type == "LOGIN_FAIL").
• .next("B") — sự kiện B phải đến ngay sau A (liền kề, strict contiguity), không được có sự kiện khác chen giữa.
• .followedBy("B") — B đến sau A nhưng cho phép sự kiện khác chen giữa (relaxed contiguity). Đây là dạng hay dùng nhất cho luật gian lận (giữa "đăng nhập sai" và "rút tiền" có thể có sự kiện khác).
• .followedByAny(...) — nới lỏng hơn nữa; .times(3) — lặp N lần (3 lần đăng nhập sai); .where(...) — điều kiện lọc; .within(Time.minutes(10)) — toàn bộ mẫu phải hoàn tất trong cửa sổ thời gian.

Kết quả là một luồng các "match" (mẫu khớp), mỗi match kèm các sự kiện thành phần để phát cảnh báo. Minh hoạ (Java, KHÔNG phải SQL chạy được — chỉ để hình dung API):

// Minh hoạ Flink CEP (DataStream API) — không chạy trong sandbox
Pattern<Event, ?> takeover = Pattern.<Event>begin("fails")
    .where(e -> e.type.equals("LOGIN_FAIL"))
    .times(3)                              // 3 lần đăng nhập sai
    .followedBy("withdraw")                // rồi (cho phép sự kiện chen giữa)
    .where(e -> e.type.equals("WITHDRAW") && e.amount > 100_000_000)
    .within(Time.minutes(10));             // trong vòng 10 phút

CEP.pattern(loginAndTxnStream.keyBy(e -> e.accountId), takeover)
   .select(match -> buildAlert(match));    // phát cảnh báo khi khớp

Lưu ý: mẫu chạy theo key (keyBy(accountId)) nên mỗi tài khoản có trạng thái mẫu riêng, và mẫu dùng event-time + within nên phụ thuộc watermark (Flink 3) để xử lý sự kiện đến trễ đúng đắn.

MATCH_RECOGNIZE (Flink SQL)

Cùng khả năng nhưng diễn đạt bằng SQL chuẩn (SQL:2016) qua MATCH_RECOGNIZE, đã giới thiệu ở Flink 5. Cú pháp PATTERN (A B+ C) dùng ký hiệu kiểu regex trên dòng sự kiện đã sắp theo thời gian; DEFINE khai báo điều kiện từng ký hiệu, MEASURES trích giá trị ra. Minh hoạ (Flink SQL — KHÔNG chạy trong sandbox PostgreSQL):

-- Minh hoạ Flink SQL (MATCH_RECOGNIZE) — KHÔNG chạy được trong sandbox
SELECT account_id, start_ts, big_amount
FROM txn_stream
MATCH_RECOGNIZE (
  PARTITION BY account_id
  ORDER BY event_time
  MEASURES FIRST(F.event_time) AS start_ts, W.amount AS big_amount
  PATTERN (F{3} W) WITHIN INTERVAL '10' MINUTE
  DEFINE
    F AS F.type = 'LOGIN_FAIL',
    W AS W.type = 'WITHDRAW' AND W.amount > 100000000
);

Chọn cái nào? Dùng MATCH_RECOGNIZE khi luật diễn đạt được bằng SQL và đội thiên về analyst (self-service, dễ đọc). Dùng CEP library khi cần logic điều kiện phức tạp bằng code Java, tích hợp mô hình, hoặc mẫu mà SQL khó biểu diễn. Cả hai chạy trên cùng runtime Flink.

Use case 4 — Phân tích real-time & dashboard

Không phải mọi thứ đều là chặn gian lận. Vận hành cần số liệu sống: tổng giá trị giao dịch theo phút, số giao dịch lỗi/từ chối, phân bố theo kênh (ATM/POS/Internet Banking), top merchant. Flink làm phần tổng hợp theo cửa sổ (đếm/sum/count-distinct theo TUMBLE/HOP) rồi sink kết quả ra ClickHouse làm kho phân tích cho dashboard (xem ClickHouse — tổng quan) hoặc ra Kafka topic cho hệ khác tiêu thụ. So với "dashboard chạy query batch trên DWH": số liệu chỉ trễ vài giây thay vì hàng giờ, và Flink đã tiền tổng hợp nên tải xuống kho phân tích nhẹ.

Use case 5 — Làm giàu luồng (stream enrichment)

Luồng giao dịch thô thường thiếu ngữ cảnh: chỉ có account_id, amount, merchant_code. Để chấm điểm hay báo cáo cần enrich thêm tên/hạng khách hàng, ngành nghề merchant, tỷ giá quy đổi. Flink có hai kiểu:

• Lookup join / async I/O: với mỗi giao dịch, tra một bảng ngoài (Redis, HBase, JDBC) lấy thuộc tính. Dùng async I/O để không chặn pipeline khi DB chậm.
• Temporal join (join theo thời điểm): join giao dịch với phiên bản đúng-tại-thời-điểm của một bảng thay đổi chậm — ví dụ bảng tỷ giá. Giao dịch lúc 10h05 phải nhân với tỷ giá có hiệu lực lúc 10h05, không phải tỷ giá mới nhất. Đây là điểm mạnh riêng của stream processing.

Use case 6 — Real-time feature cho ML / feature store

Mô hình chống gian lận chỉ tốt bằng feature nó nhận. Nhiều feature mạnh nhất là real-time và stateful: "số giao dịch trong 5 phút", "tỷ lệ giao dịch bị từ chối 1 giờ qua", "khoảng cách địa lý so với giao dịch trước". Flink tính các feature này trên luồng rồi đẩy vào feature store online (Redis/low-latency store) để mô hình serving đọc lúc chấm điểm, đồng thời ghi bản sao ra kho offline để train — đảm bảo train/serving dùng cùng một công thức feature (tránh training/serving skew). Nghiệp vụ giám sát và feature chống gian lận xem thêm AML — giám sát giao dịch.

Kiến trúc tham chiếu

Gom mọi use case lại thành một kiến trúc thống nhất mà NCB có thể dùng:

Một luồng vào (Kafka), Flink chia nhánh xử lý (enrich → CEP / aggregation / scoring), và ba loại đầu ra: cảnh báo (chặn/điều tra), phân tích (ClickHouse/dashboard), feature (feature store). Kafka làm lớp đệm chịu tải và tách rời nguồn với xử lý (xem Kafka — kiến trúc).

Chọn công cụ: Flink vs Spark Streaming vs Kafka Streams

Không phải mọi bài toán streaming đều cần Flink. Bảng dưới giúp chọn:

Nguyên tắc:

• Cần chặn gian lận độ trễ mili giây + CEP mẫu chuỗi + state lớn → Flink.
• Đã đầu tư Spark, cần ETL/analytics gần real-time, chấp nhận trễ vài giây, muốn dùng chung code batch → Spark Structured Streaming.
• Chỉ cần một micro-service nhỏ biến đổi/lọc/join giữa các topic Kafka, không muốn dựng cụm → Kafka Streams (nhẹ, nhúng thẳng vào ứng dụng).

Cạm bẫy & lưu ý sản xuất

• Nghĩ CEP giải quyết mọi thứ. Mẫu quá rộng (followedByAny, cửa sổ dài) làm state phình và tạo nhiều match giả. Bắt đầu chặt (strict next, cửa sổ ngắn), nới dần theo dữ liệu.
• Bỏ qua sự kiện đến trễ. Fraud thường dựa event-time; sự kiện trễ mà watermark đã qua sẽ rơi khỏi mẫu, bỏ lọt gian lận. Cấu hình allowed lateness/side output hợp lý.
• State không có TTL. Keyed state theo tài khoản không đặt TTL sẽ phình vô hạn theo số tài khoản đã từng xuất hiện → OOM. Luôn đặt state TTL (xem Flink 4).
• Data skew (khoá nóng). Một merchant lớn chiếm phần lớn giao dịch làm một subtask nghẽn; xử lý bằng pre-aggregation/salting (Flink 7).
• Sink chậm chặn cả pipeline. Ghi cảnh báo/đặc trưng vào DB đồng bộ gây backpressure; dùng async I/O và buffer.
• Cảnh báo dội bom (alert fatigue). Ngưỡng quá nhạy tạo hàng nghìn cảnh báo/ngày, đội điều tra bỏ qua. Cân bằng precision/recall, có cơ chế dedupe và gộp cảnh báo cùng tài khoản.
• Quên tương thích state khi nâng cấp luật. Đổi mẫu/state mà không gán UID tường minh → mất state khi deploy (Flink 7).

Đối chiếu: velocity check dạng batch (chạy được)

Để cảm nhận một velocity rule trông thế nào, dưới đây là phiên bản batch trên sandbox PostgreSQL: đếm giao dịch mỗi tài khoản trong một khoảng 10 phút bất kỳ và lọc ra tài khoản vượt ngưỡng. Trong Flink, cùng logic này chạy liên tục theo cửa sổ trượt trên luồng; ở đây là ảnh chụp tĩnh để đối chiếu ý tưởng ngưỡng.

-- ▶ Chạy được
WITH per_10min AS (
  SELECT
    account_id,
    date_trunc('hour', created_at)
      + floor(extract(minute FROM created_at) / 10) * interval '10 minute' AS bucket,
    COUNT(*)      AS txn_count,
    SUM(amount)   AS total_amount
  FROM transactions
  GROUP BY 1, 2
)
SELECT account_id, bucket, txn_count, total_amount
FROM per_10min
WHERE txn_count >= 20 OR total_amount >= 500000000
ORDER BY txn_count DESC;

Câu này gom giao dịch vào các "xô" 10 phút, đếm và cộng tiền theo tài khoản, rồi lọc xô nào vượt ngưỡng (>=20 giao dịch hoặc >=500 triệu). Đó chính là bản chất velocity check — chỉ khác là Flink làm nó trên cửa sổ trượt real-time thay vì quét bảng một lần.

Use case thực tế

Bối cảnh. NCB triển khai một pipeline chống chiếm đoạt tài khoản (account takeover) trên Flink: Kafka gom cả sự kiện đăng nhập (từ Internet Banking) lẫn giao dịch (từ core banking) vào chung một luồng, keyBy theo account_id.

Luật CEP. Đội rủi ro cài mẫu: 3 lần đăng nhập sai rồi (followedBy, cho phép sự kiện chen giữa) một lệnh chuyển/rút > 100 triệu, tất cả trong vòng 10 phút (event-time). Khớp mẫu → phát cảnh báo ra topic fraud-alerts mà hệ thống chặn đang lắng nghe.

Số liệu vận hành. Cụm xử lý ~4.000 sự kiện/giây giờ thường. Độ trễ đầu-cuối từ lúc lệnh rút vào Kafka đến lúc cảnh báo phát ra ~180 ms — đủ để chặn đồng bộ trước khi tiền rời tài khoản. State giữ mẫu đang dở của mỗi tài khoản có TTL 15 phút nên không phình. Trong một tháng, luật bắt được 37 vụ tiếp quản tài khoản với ~2% cảnh báo giả sau khi đã siết ngưỡng và gộp cảnh báo trùng.

Nhánh phân tích song song. Cùng luồng đó, một job aggregation đếm tỷ lệ đăng nhập sai theo phút và ghi ra ClickHouse; đội SOC nhìn dashboard thấy tỷ lệ đăng nhập sai toàn hệ thống tăng đột biến (dấu hiệu tấn công dò mật khẩu diện rộng) và chủ động nâng ngưỡng, bật thêm xác thực — thứ mà một job chặn đơn lẻ không thấy được.

Ghi nhớ

• Flink thắng khi dữ liệu đáng giá nhất lúc đang xảy ra: fraud real-time, AML velocity/structuring, CEP mẫu chuỗi — độ trễ mili giây, stateful theo tài khoản.
• CEP = phát hiện MẪU CHUỖI sự kiện, không phải tổng hợp. Hai cách: Flink CEP library (begin → next/followedBy → within, .times, .where) cho code Java; MATCH_RECOGNIZE (PATTERN/DEFINE/MEASURES) cho SQL. next = liền kề, followedBy = cho phép chen giữa.
• Các use case bổ trợ: phân tích/dashboard (window → ClickHouse), enrich (lookup/temporal join tỷ giá), real-time feature cho ML (dùng chung công thức train/serving).
• Kiến trúc tham chiếu: Kafka → Flink (enrich → CEP/aggregation/scoring) → cảnh báo + phân tích + feature.
• Chọn công cụ: Flink cho fraud/CEP/state lớn độ trễ thấp; Spark cho ETL/analytics gần real-time đã có sẵn Spark; Kafka Streams cho micro-service gọn quanh Kafka.
• Cạm bẫy: mẫu CEP quá rộng, bỏ qua sự kiện trễ (watermark), state không TTL, khoá nóng, sink chậm gây backpressure, cảnh báo dội bom, quên UID khi nâng cấp.
• Ngưỡng velocity là logic đơn giản (đếm/sum vượt ngưỡng theo cửa sổ); giá trị của Flink là làm nó liên tục, real-time, có context state thay vì batch cuối ngày.