Flink 7 — Triển khai, mở rộng & vận hành

Từ job chạy được sang cụm chạy 24/7

Ở các bài trước, ta đã có một job Flink đúng logic: đọc luồng giao dịch, tính cửa sổ event-time, giữ state, checkpoint để không mất dữ liệu khi sập. Nhưng "job chạy đúng trên laptop" và "cụm giám sát giao dịch chạy liên tục nhiều tháng, tự phục hồi, mở rộng theo tải, nâng cấp không mất state" là hai thế giới khác nhau. Bài này nói về thế giới thứ hai: vận hành Flink trong production.

Với một ngân hàng, đây không phải chuyện lý thuyết. Cụm Flink giám sát giao dịch phải sống 24/7 — mỗi phút chết là mỗi phút giao dịch gian lận có thể lọt lưới. Nó phải chịu được node hỏng, chịu được tải tăng vọt giờ cao điểm, và cho phép đội phát triển vá luật phát hiện gian lận mới mà không đánh mất state đã tích lũy (ví dụ hồ sơ hành vi khách hàng trong 30 ngày). Ta sẽ đi qua: chọn chế độ triển khai, chọn nền tảng chạy, cấu hình tài nguyên, mở rộng, nâng cấp an toàn, giám sát, high availability và các sự cố hay gặp.

Ba chế độ triển khai

Trước khi chọn chạy Flink ở đâu, phải hiểu cách một job được đóng gói và ánh xạ lên cluster. Flink có ba chế độ, khác nhau ở chỗ ai chạy main() của ứng dụng và cluster có được chia sẻ hay không.

Session mode (chế độ phiên). Bạn khởi động sẵn một Flink cluster chạy dài hạn, chưa có job nào, rồi submit nhiều job vào cùng cluster đó. Client chạy main(), dựng ra job graph rồi gửi lên JobManager. Ưu điểm: cluster khởi động một lần, submit job mới rất nhanh (không phải bật cluster mới), tiết kiệm tài nguyên cho nhiều job nhỏ. Nhược điểm nghiêm trọng: cách ly kém. Các job chia sẻ chung TaskManager; một job ngốn bộ nhớ gây OutOfMemory có thể kéo TaskManager sập, làm tất cả các job trên đó chết theo. Một job nặng gây backpressure cũng ảnh hưởng hàng xóm. Session mode hợp cho môi trường dev/test hoặc nhiều job ad-hoc ngắn, không khuyến nghị cho job production quan trọng.

Per-job mode (chế độ theo job — đã cũ/deprecated). Mỗi job có một cluster riêng: khi submit, Flink dựng một cluster mới chỉ để chạy đúng job đó, job xong thì cluster dọn đi. Cách ly tốt hơn session mode. Nhưng client vẫn chạy main() để dựng job graph rồi mới gửi lên — điều này khiến việc đóng gói và triển khai nặng nề hơn. Chế độ này đã bị đánh dấu deprecated và đang được thay thế bằng application mode; nhắc đến ở đây để bạn nhận ra nó trong tài liệu cũ.

Application mode (chế độ ứng dụng — khuyến nghị cho production). Mỗi ứng dụng (application) có một cluster riêng dành riêng cho nó, và điểm mấu chốt: main() của ứng dụng chạy ngay trên JobManager, không phải trên client. Client chỉ gửi JAR + cấu hình, còn việc dựng job graph diễn ra bên trong cluster. Lợi ích:

• Cách ly hoàn toàn: mỗi application có JobManager và TaskManager riêng, không job nào ảnh hưởng job khác về tài nguyên hay lỗi.
• Client nhẹ, không còn là điểm nghẽn: không cần client tải và biên dịch dependency nặng, không lo client mất kết nối làm hỏng submit.
• Vòng đời rõ ràng: cluster sống và chết cùng ứng dụng — rất hợp mô hình khai báo trên Kubernetes.

Nguyên tắc thực chiến: production dùng application mode, mỗi job giám sát giao dịch quan trọng là một application độc lập.

Nền tảng chạy: standalone, YARN, Kubernetes

Chế độ triển khai trả lời "job ánh xạ lên cluster ra sao"; nền tảng chạy trả lời "cluster đó sống trên hạ tầng gì".

Standalone. Bạn tự cài JobManager và các TaskManager như tiến trình JVM trên máy (bare-metal hoặc VM), tự quản lý khởi động, khởi động lại. Đơn giản, không phụ thuộc resource manager, tốt để học và thử nghiệm. Nhưng thiếu tự động cấp phát tài nguyên và tự phục hồi ở mức hạ tầng — bạn phải tự lo mọi thứ. Ít dùng cho production quy mô lớn.

YARN. Chạy Flink trên Hadoop YARN, để YARN cấp phát container cho JobManager/TaskManager. Phù hợp với ngân hàng đã có sẵn cụm Hadoop và muốn Flink dùng chung tài nguyên với Spark, Hive. YARN lo lập lịch, cách ly container, khởi động lại khi container chết. Vẫn phổ biến ở các hệ thống on-prem có lịch sử Hadoop.

Kubernetes (khuyến nghị cho hạ tầng hiện đại). Flink chạy native trên Kubernetes: JobManager và TaskManager là các Pod, Flink tự yêu cầu K8s cấp thêm TaskManager Pod khi cần. Kết hợp mượt với mô hình khai báo của K8s và với application mode. Với đội đã vận hành K8s (xem Kubernetes — vận hành production), đây là lựa chọn tự nhiên nhất.

Flink Kubernetes Operator

Chạy Flink native trên K8s vẫn còn thủ công. Flink Kubernetes Operator nâng lên một tầng: nó đưa vào K8s các custom resource để bạn khai báo ứng dụng Flink, còn Operator lo toàn bộ vòng đời — dựng cluster, submit job, chụp savepoint, nâng cấp, khôi phục HA. Bạn mô tả trạng thái mong muốn, Operator lo đưa hệ thống về đúng trạng thái đó (đúng triết lý declarative của K8s).

Resource chính là FlinkDeployment. Ví dụ minh hoạ (YAML mô tả, không phải SQL chạy được):

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: fraud-monitoring
spec:
  image: registry.ncb/flink-fraud:1.4.0
  flinkVersion: v1_18
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
    state.backend: rocksdb
    execution.checkpointing.interval: "30s"
    high-availability.type: kubernetes
    state.checkpoints.dir: s3://ncb-flink/checkpoints
    state.savepoints.dir: s3://ncb-flink/savepoints
  serviceAccount: flink
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "4096m"
      cpu: 2
  job:
    jarURI: local:///opt/flink/usrlib/fraud-job.jar
    parallelism: 8
    upgradeMode: savepoint   # nâng cấp qua savepoint, không mất state
    state: running

Điểm đáng chú ý: upgradeMode: savepoint khiến Operator tự chụp savepoint trước khi nâng cấp và khôi phục từ đó khi bạn đổi image — đây là chìa khoá để vá luật gian lận mà không đánh mất state (mục nâng cấp bên dưới).

Cấu hình tài nguyên: slot và bộ nhớ

Sau khi chọn nơi chạy, phải cấp đúng tài nguyên. Sai ở đây thì job hoặc lãng phí máy, hoặc OutOfMemory lúc nửa đêm.

Task slot. Đơn vị tài nguyên của TaskManager. Mỗi TaskManager có một số slot cố định (taskmanager.numberOfTaskSlots); mỗi slot chạy một slice song song của pipeline. Số slot khả dụng của cả cluster = số TaskManager × slot mỗi TaskManager, và đây chính là giới hạn trần cho parallelism của job.

Parallelism. Số bản sao song song của một toán tử. Job parallelism 8 nghĩa mỗi toán tử (mặc định) chạy 8 subtask song song, cần tối thiểu 8 slot. Parallelism là cần gạt chính để tăng thông lượng — nhiều subtask hơn xử lý nhiều phân vùng dữ liệu hơn cùng lúc.

Slot sharing. Theo mặc định, các subtask từ các toán tử khác nhau nhưng cùng một pipeline có thể chia sẻ cùng một slot. Nghĩa là một slot có thể chứa "một lát cắt dọc" đầy đủ của pipeline (source → map → window → sink). Lợi ích: một job parallelism 8 chỉ cần 8 slot dù có nhiều toán tử, và tải phân bổ đều hơn giữa các slot. Bạn có thể tách bằng "slot sharing group" khi cần cô lập toán tử nặng.

Mô hình bộ nhớ TaskManager. Bộ nhớ một TaskManager không chỉ là "heap". Flink chia tổng bộ nhớ tiến trình thành nhiều vùng, và hiểu chúng là chìa khoá gỡ lỗi OOM:

• Framework + Task Heap: heap JVM cho code toán tử của bạn (biến, đối tượng người dùng tạo).
• Managed memory (bộ nhớ quản lý): vùng off-heap Flink tự quản, dùng cho RocksDB state backend và cho việc sort/hash trong batch. Job state lớn (như hồ sơ hành vi khách) cần managed memory rộng — thiếu là RocksDB đọc/ghi đĩa nhiều, chậm hẳn.
• Network memory (bộ nhớ mạng): các network buffer để truyền dữ liệu giữa các subtask (shuffle giữa các TaskManager). Thiếu buffer gây lỗi "insufficient network buffers" hoặc làm backpressure lan nhanh.
• JVM Metaspace + Overhead: metaspace cho class, cộng phần đệm cho stack thread, GC, native.

Nguyên tắc: job state lớn → tăng managed memory; job shuffle nhiều → chú ý network memory; đừng chỉ đôn heap lên rồi thắc mắc sao vẫn OOM. Cấu hình gọn nhất là đặt taskmanager.memory.process.size (tổng) rồi để Flink chia theo tỉ lệ, tinh chỉnh từng vùng khi profiling chỉ ra nút thắt.

Mở rộng và rescaling

Tải giao dịch không cố định: giờ cao điểm trưa và cuối tháng có thể gấp nhiều lần ban đêm. Cần đổi được parallelism.

Rescaling qua savepoint (cách chuẩn, thủ công). Parallelism không thể đổi khi job đang chạy theo cách "nóng" tuỳ tiện, vì state được phân vùng theo parallelism. Cách an toàn là qua savepoint:

1. stop-with-savepoint: dừng job và chụp một savepoint nhất quán (toàn bộ state được lưu ra object store).
2. Khởi động lại job từ savepoint đó với parallelism mới. Flink phân phối lại (redistribute) state theo parallelism mới nhờ cơ chế key-group — state của các key được chia lại cho số subtask mới mà không mất mát.

Đây cũng chính là quy trình được các nền tảng như Operator tự động hoá.

Reactive mode / autoscaling. Flink hỗ trợ reactive mode: job tự động dùng hết số slot đang có — bạn chỉ cần thêm/bớt TaskManager, job tự đổi parallelism theo. Kết hợp với autoscaler (Flink Kubernetes Operator có autoscaler tích hợp) để tự động co giãn theo tải: khi backpressure/lag Kafka tăng, autoscaler chụp savepoint và khởi động lại với parallelism cao hơn; tải giảm thì thu nhỏ. Với cụm giám sát giao dịch, đây là cách xử lý spike giờ cao điểm mà không phải trực tay.

Nâng cấp job an toàn bằng savepoint

Đây là kỹ năng vận hành quan trọng nhất và cũng dễ làm hỏng nhất. Đội gian lận muốn thêm luật mới, đổi ngưỡng, sửa bug — tức đổi code — nhưng không được mất state (mất hồ sơ 30 ngày là mù cả tháng).

Savepoint là ảnh chụp state toàn vẹn do người vận hành chủ động kích hoạt (khác checkpoint tự động của Flink 6 — Fault tolerance, vốn để tự phục hồi). Quy trình nâng cấp:

Mấu chốt là tương thích state (state compatibility). State được ánh xạ lại theo:

• Operator UID: Flink khớp state của toán tử cũ với toán tử mới theo UID do bạn đặt. Vì vậy luôn gán UID tường minh cho các toán tử có state (.uid("fraud-profile")); nếu để Flink tự sinh UID theo cấu trúc graph, chỉ cần sửa nhẹ pipeline là UID đổi và không nạp lại được state cũ.
• Schema state: nếu bạn đổi kiểu dữ liệu của state (thêm/bớt field), cần dùng serializer hỗ trợ schema evolution (ví dụ Avro/POJO theo quy tắc của Flink). Đổi kiểu tuỳ tiện có thể khiến state không tương thích.

Với upgradeMode: savepoint ở FlinkDeployment, Operator tự làm toàn bộ chuỗi stop-with-savepoint → deploy image mới → restore. Việc của bạn chỉ là giữ UID ổn định và đảm bảo schema tương thích.

Giám sát cụm

Không có giám sát thì "chạy 24/7" chỉ là hy vọng.

Flink Web UI. Giao diện tích hợp trên JobManager, nơi nhìn thấy sức khoẻ job trực quan:

• Job graph: sơ đồ toán tử, thông lượng (records/s) từng toán tử.
• Backpressure: đánh dấu toán tử nào đang bị nghẽn (màu đỏ) — chỉ thẳng nút thắt.
• Checkpoint: lịch sử checkpoint, thời gian hoàn thành, kích thước, số lần fail — dấu hiệu sớm nhất của trục trặc state.
• Watermark: watermark hiện tại từng subtask — phát hiện subtask bị "kẹt" watermark làm cửa sổ không đóng.

Metrics ra Prometheus. Web UI tốt để soi tức thời nhưng không thay được giám sát dài hạn và cảnh báo. Flink expose metrics qua metric reporter Prometheus; Prometheus scrape rồi Grafana dựng dashboard, Alertmanager bắn cảnh báo. Các metric cần theo dõi: Kafka consumer lag (job có theo kịp luồng không), checkpoint duration & failure count, backpressure/busyTime, số lần restart, heap/managed memory usage. Xem thêm Observability — tổng quan cho cách dựng tầng giám sát chuẩn.

Log. Log JobManager/TaskManager (thường gom về ELK/Loki) là nơi cuối cùng để truy nguyên: exception gây restart, cảnh báo GC dài, lỗi kết nối sink. Với K8s, log ra stdout của Pod và gom tập trung.

Xử lý sự cố thường gặp

Nguyên tắc chung: backpressure là triệu chứng, không phải bệnh — luôn lần ngược từ toán tử đỏ về nút thắt thực (thường là sink hoặc một hàm nặng), và nhớ rằng checkpoint fail thường là hệ quả của backpressure chứ không phải lỗi riêng.

High availability (HA)

JobManager là điểm điều phối trung tâm — nếu nó chết mà không có HA, job dừng và mất metadata điều phối. HA cho JobManager giải quyết bằng cách lưu metadata quan trọng (job graph, con trỏ tới checkpoint mới nhất, thông tin bầu leader) ra một kho bền vững, để một JobManager dự phòng có thể tiếp quản:

• ZooKeeper HA: dùng ZooKeeper để bầu leader và lưu con trỏ; metadata lớn để trên HDFS/S3. Phổ biến trong hạ tầng standalone/YARN.
• Kubernetes HA: dùng chính cơ chế của K8s (ConfigMap + lease để bầu leader), lưu con trỏ checkpoint ra object store — không cần ZooKeeper. Đây là lựa chọn gọn nhất khi đã chạy trên K8s; đúng cấu hình high-availability.type: kubernetes trong ví dụ FlinkDeployment ở trên.

Lưu ý: HA bảo vệ khả năng điều phối và khôi phục, còn dữ liệu state được bảo vệ bởi checkpoint. Hai cơ chế bổ trợ nhau: khi JobManager mới lên, nó đọc con trỏ HA để biết checkpoint gần nhất rồi khôi phục state từ đó.

Use case thực tế

Bối cảnh. Cụm Flink giám sát giao dịch của NCB chạy trên Kubernetes qua Flink Kubernetes Operator, xử lý ~4.000 giao dịch/giây giờ thường, tăng lên ~12.000/giây trưa và cuối tháng. Job giữ state hồ sơ hành vi 30 ngày cho mỗi tài khoản, checkpoint mỗi 30 giây ra S3.

Tình huống 1 — spike cuối tháng. Kafka consumer lag bắt đầu tăng lúc 11h30 cuối tháng. Dashboard Grafana bắn cảnh báo khi lag vượt 60 giây. Autoscaler của Operator phát hiện backpressure kéo dài, tự chụp savepoint và khởi động lại job với parallelism 8 → 16 (thêm TaskManager Pod). State được phân phối lại theo key-group; sau ~90 giây job bắt kịp, lag về gần 0. Chiều tối tải giảm, autoscaler thu về parallelism 8, tiết kiệm 8 Pod.

Tình huống 2 — nâng cấp luật gian lận. Đội rủi ro thêm một luật phát hiện "giao dịch bất thường theo vị trí" và sửa ngưỡng một luật cũ. Vì các toán tử có state đều đã gán UID tường minh (fraud-profile, velocity-window) và schema state không đổi kiểu, kỹ sư chỉ đổi image trong FlinkDeployment và kubectl apply. Operator (đang ở upgradeMode: savepoint) tự stop-with-savepoint job v1.4, deploy v1.5, restore từ savepoint. Không mất một byte state 30 ngày nào; tổng gián đoạn ~40 giây, ngoài giờ cao điểm.

Tình huống 3 — data skew. Một tài khoản merchant lớn tạo ~15% tổng giao dịch, làm một subtask nóng gấp nhiều lần các subtask khác, gây backpressure cục bộ dù cả cụm còn rảnh. Đội thêm bước local pre-aggregation trước keyBy và salting cho key nóng; tải phân bổ đều lại, backpressure biến mất mà không cần thêm tài nguyên.

Ghi nhớ

• Production dùng application mode (main() chạy trên JobManager, cách ly tốt nhất); session mode chỉ cho dev/nhiều job ad-hoc; per-job đã deprecated.
• Nền tảng: standalone (đơn giản), YARN (đã có Hadoop), Kubernetes qua Flink Operator (khai báo FlinkDeployment, tự động hoá vòng đời) cho hạ tầng hiện đại.
• Tài nguyên: slot là đơn vị của TaskManager và là trần của parallelism; slot sharing cho phép một slot chứa cả pipeline; hiểu mô hình bộ nhớ (heap / managed cho RocksDB / network buffer) để gỡ OOM đúng chỗ.
• Rescaling an toàn qua stop-with-savepoint → restart với parallelism mới (state redistribute theo key-group); reactive mode + autoscaler để co giãn theo tải tự động.
• Nâng cấp không mất state: savepoint + UID tường minh + schema tương thích. Luôn gán UID cho toán tử có state.
• Giám sát: Web UI (job graph, backpressure, checkpoint, watermark) để soi tức thời; metrics → Prometheus/Grafana cho cảnh báo (Kafka lag, checkpoint duration, restart count); log tập trung để truy nguyên.
• Sự cố hay gặp: checkpoint fail và backpressure thường liên đới; state phình do thiếu TTL; skew khoá xử lý bằng salting/pre-aggregation.
• HA JobManager qua ZooKeeper hoặc Kubernetes (ConfigMap + lease) lưu metadata điều phối; HA bổ trợ checkpoint chứ không thay thế.