Hành trình áp dụng Flink SQL

• Hyperconnect Azar Matching Dev Team đã chọn cách triển khai xử lý streaming bằng SQL thay vì code ứng dụng, nhằm tách ứng dụng legacy Flink nguyên khối đang dùng tới 96 CPU
• Chia thành nhiều Flink App sẽ cải thiện tính cô lập nhưng làm tăng gánh nặng vận hành, nên nhóm đánh giá Flink SQL phù hợp hơn về năng suất và hiệu quả vận hành nhờ có thể tận dụng mức độ hiểu biết nội bộ về Flink của đội
• Flink SQL hỗ trợ Checkpoint/Savepoint, JobManager HA, phân phối lại TaskManager, window·join·event time·watermark, UDF và custom connector, nên phù hợp với bối cảnh của nhóm hơn ksqlDB và Spark Structured Streaming
• Môi trường vận hành được xây dựng dưới dạng Session mode Flink Cluster trên Kubernetes, và dùng Flink SQL Gateway API cùng GitHub Actions để triển khai truy vấn và dừng Job theo mô hình GitOps
• Hệ thống đã vận hành ổn định gần 1 năm và đang tiếp tục mở rộng, nhưng việc triển khai lại truy vấn và thay đổi cấu hình cluster vẫn còn bất tiện, nên nhóm dự định cải tiến dựa trên mẫu GitOps Controller

---

Bối cảnh thay thế ứng dụng streaming legacy nặng nề

• Azar Matching Dev Team đang quản lý nhiều ứng dụng dựa trên Flink, trong đó có một ứng dụng legacy nặng dùng tới 96 CPU
• Ứng dụng này có kiến trúc nguyên khối, gom nhiều chức năng vào một chỗ như join nhiều sự kiện matching, phát hành sự kiện có điều kiện, và lưu cờ trong Redis
• Sau khi thay đổi node chạy do công việc hạ tầng toàn công ty, ứng dụng không còn hoạt động bình thường, và rất khó giải quyết nhanh chỉ bằng tinh chỉnh đơn giản
• Vì chức năng join sự kiện quan trọng đã được triển khai sẵn trong một Flink app mới của dự án khác, nhóm cần một cách thay thế phần phát hành sự kiện có điều kiện và logic xử lý sau bước join sự kiện

So sánh các cách thay thế

• Nếu triển khai bằng một Flink App duy nhất thì số đối tượng phải quản lý sẽ ít, nhưng rất dễ lại trở thành một ứng dụng khổng lồ, và lỗi ở một phần có thể ảnh hưởng tới các chức năng khác
• Nếu chia thành nhiều Flink App thì có thể quản lý từng app độc lập, nhưng số app càng tăng thì gánh nặng về cluster, tài nguyên và triển khai cũng tăng theo
• Flink SQL cho phép định nghĩa logic bằng truy vấn để phát triển nhanh và chỉ cần quản lý một cluster, nhưng khó biểu diễn logic phức tạp chỉ bằng SQL, đồng thời đòi hỏi kinh nghiệm vận hành cluster
• Nhóm khi đó đã có mức độ hiểu biết cao hơn về cách Flink được triển khai bên trong, và đánh giá Flink SQL có lợi thế về năng suất và hiệu quả vận hành

Lý do chọn Flink SQL

• Flink SQL cho phép triển khai ứng dụng xử lý event streaming bằng SQL mà không cần trực tiếp viết mã ứng dụng
• Về mặt tính sẵn sàng cao (HA), Flink hỗ trợ xử lý có trạng thái và có thể lưu/khôi phục trạng thái công việc theo chu kỳ hoặc tại thời điểm mong muốn bằng Checkpoint và Savepoint
  • JobManager có thể được cấu hình ở chế độ HA dạng leader-standby
  • Nếu một phần TaskManager bị lỗi, công việc của TaskManager đó có thể được phân phối lại sang TaskManager khác theo chiến lược retry của Job
• Chỉ với cú pháp SQL cũng có thể xử lý các chức năng streaming chính
  • Dùng SELECT để biến đổi dữ liệu và WHERE để lọc bản ghi
  • Có thể kết hợp nhiều stream bằng JOIN và gộp stream bằng UNION
  • Hỗ trợ xử lý window như tumbling, hopping(sliding), session window
  • Có thể thiết lập phạm vi chấp nhận dữ liệu đến trễ bằng event time và watermark
• Có thể mở rộng ra ngoài tính năng mặc định bằng UDF và custom connector
  • Phần lớn logic legacy hiện có dùng mẫu lệnh Redis SET hoặc INCR, nhưng do không có Redis Connector chính thức của Flink nên nhóm đã tự viết Redis Connector để dùng
  • Khi đó chưa có hàm built-in để lấy giao của kiểu ARRAY, nên nhóm đã triển khai bằng UDF và dùng trong truy vấn

So sánh với ksqlDB và Spark Structured Streaming

• ksqlDB đã có sẵn trong nền tảng Confluent mà công ty dùng cho Kafka, và cũng đã có các trường hợp sử dụng nội bộ
• Tuy nhiên, nhóm đánh giá nó kém hiệu quả về hành vi HA cho xử lý streaming có trạng thái
  • Khi failover cho tác vụ có trạng thái, cần replay toàn bộ changelog ghi lại các thay đổi state nên thời gian failover có thể kéo dài
  • Cách giữ bản sao của stream xử lý và liên tục cập nhật changelog vào state nội bộ khiến bản sao cũng phải thực hiện cùng phép tính, nên tài nguyên có thể bị tiêu tốn gấp đôi
  • Có thể xem thêm tại Configuring ksqlDB for High Availability | Confluent Developer
• Spark Structured Streaming là engine xử lý streaming dựa trên Spark SQL engine
  • Đã có trường hợp sử dụng trong công ty và có thể viết UDF cùng Custom Sink
  • Có hệ sinh thái lớn hơn và được xây dựng hoàn thiện hơn Flink
• Spark hoạt động theo đơn vị micro-batch nên có thể phát sinh độ trễ ở mức bản ghi, và trong bối cảnh cần xử lý thời gian thực thì có thể bất lợi hơn Flink
• Do nhóm hầu như không có kinh nghiệm với Spark và cũng cần viết Custom Sink, nên rất khó để mạnh dạn chọn Spark

Thiết lập môi trường cluster

• Trên local, có thể tải binary từ trang chính thức của Flink và chạy cluster bằng {FLINK_HOME}/bin/start-cluster.sh
• Khi chạy {FLINK_HOME}/bin/sql-client.sh, Flink SQL CLI sẽ mở ra và có thể gửi truy vấn thử như SELECT 1;
• Sau khi gửi truy vấn, có thể xác nhận trên Flink web UI rằng truy vấn đã được chuyển thành Job để chạy
• Cuối năm 2022, Flink SQL Gateway được phát hành, cho phép gửi truy vấn qua HTTP

Kiến trúc vận hành dựa trên Kubernetes

• Vì phần lớn dịch vụ nội bộ chạy trên Kubernetes, Flink SQL Cluster cũng được dựng trên Kubernetes
• Các Flink App hiện có trước đó đều đang được triển khai và vận hành theo Application mode
  • Đây là cách khởi tạo một cluster riêng cho từng ứng dụng
  • Trên Kubernetes, mỗi app sẽ chạy riêng JobManager Pod và TaskManager Pod
  • Cách này có lợi về tính độc lập và cô lập giữa các app, cũng như về cấu hình và quản lý phụ thuộc theo từng công việc
• Vì Flink SQL gửi Job vào một cluster đã chạy sẵn, JobManager và TaskManager cần được chạy ở Session mode
• Cluster được cấu hình dựa trên hướng dẫn Stand Alone Cluster on Kubernetes
• Môi trường HA tham khảo cấu hình High-Availability with Standalone Kubernetes, và dùng s3 cho high-availability.storageDir
• Nhóm đánh giá cách Native Kubernetes không phù hợp với hạ tầng nội bộ, vì nó khởi tạo cluster bằng shell script được cung cấp sẵn thay vì tự định nghĩa deployment để triển khai

Cấu hình HA và tích hợp S3

• Để cấu hình HA và tích hợp S3, nhóm dùng các thiết lập sau trong config.yaml

high-availability.type: kubernetes
high-availability.storageDir: s3://{s3-path-for-flinksql-recovery}
kubernetes.cluster-id: {cluster-id}
kubernetes.namespace: {k8s-namespace}


# namespace 내의 service account 를 통해 Kubernetes cluster 에 접근할 수 있도록 권한을 부여하는 작업이 필요할 수 있습니다.
kubernetes.service-account: {k8s-service-account-for-flinksql}

• Trong môi trường HA, sẽ chạy hai JobManager pod, và địa chỉ của chúng phải khác nhau để logic bầu leader hoạt động bình thường
• Tham số khởi chạy container JobManager được cấu hình như sau

args: ["start-foreground", "-D", "jobmanager.rpc.address=$(POD_IP)"]

• Với thiết lập này, thông tin JobManager pod hiện đang được bầu làm leader và Job ID đang chạy sẽ được lưu trong Kubernetes ConfigMap để phục vụ HA

Triển khai truy vấn theo mô hình GitOps

• Flink hiện vẫn chưa có web UI hay công cụ chuyên dụng do chính Flink cung cấp cho Flink SQL
• Nhóm từng xem xét PoC tích hợp Hue, nhưng khi đó có vấn đề tương thích phiên bản với Flink SQL Gateway, cần phát triển thêm và cũng tốn thời gian dựng môi trường phát triển
• Vì nội bộ công ty sử dụng nhiều mô hình GitOps, nhóm đã triển khai GitHub Actions để deploy truy vấn hoặc dừng Job
• Trong repository, mỗi Job có một thư mục riêng và các truy vấn cần chạy được đặt thành file SQL
• GitHub Actions nhận tên thư mục để xác định file SQL cần trích xuất truy vấn
• Việc triển khai được thực hiện bằng cách gọi Flink SQL Gateway REST API, và được viết bằng Python để đơn giản, dễ kiểm thử

Trường hợp vận hành và ứng phó sự cố

• Nhóm chưa từng gặp trường hợp JobManager lỗi, nhưng theo cấu hình HA, nếu JobManager lỗi thì JobManager còn lại có thể được bầu làm leader và tiếp tục công việc
• TaskManager thỉnh thoảng bị lỗi, phần lớn là các trường hợp Pod bị restart do chính sách QoS của Kubernetes
• Nhóm đã xác nhận rằng dù một số TaskManager lỗi, công việc vẫn tiếp tục nhờ được phân phối lại sang TaskManager khác
• Các lỗi truy vấn phần lớn xảy ra do dữ liệu bất thường được đưa vào hoặc thiếu tài nguyên tính toán
  • Khi đọc dữ liệu JSON, định dạng JSON sai có thể được bỏ qua bằng tùy chọn json.ignore-parse-errors
  • Khi dùng JSON_VALUE để trích dữ liệu từ một path cụ thể, lỗi phát sinh do không có giá trị hoặc sai kiểu có thể được xử lý bằng cách đặt giá trị mặc định với DEFAULT {VALUE} ON ERROR
  • Nếu CPU của TaskManager vượt 100% hoặc thiếu bộ nhớ, nhóm sẽ tăng tài nguyên TaskManager hoặc tăng parallelism của truy vấn rồi triển khai lại
• Khi khởi động lại cluster do thay đổi cấu hình cluster hoặc thêm UDF, đôi khi có Job bị lỗi
  • Nguyên nhân thường là thiết lập timeout hoặc retry của Job chưa phù hợp
  • Nhóm đã chỉnh timeout và retry để Job không kết thúc retry quá sớm, mà tiếp tục thử lại cho đến khi cluster ổn định sau khi restart

Ràng buộc khi thay đổi truy vấn và khôi phục state

• Khi chỉnh điều kiện truy vấn rồi deploy lại, việc khôi phục state bằng savepoint chỉ khả thi với các thay đổi rất đơn giản như sửa giá trị trong biểu thức điều kiện
• Nếu điều kiện window thay đổi thì state cũng thay đổi, khiến khó giữ tương thích và có thể không thể khôi phục bằng savepoint
• Nếu cần giữ state nhưng yêu cầu thay đổi thường xuyên, cách tự viết app có thể phù hợp hơn

Các điểm cần theo dõi khi giám sát

• Flink có rất nhiều metric tích hợp sẵn, nên nếu có hạ tầng giám sát nội bộ và Metric Reporter phù hợp thì có thể dễ dàng xây dựng môi trường monitoring
• numRunningJobs cho biết số Job đang chạy hiện tại trong cluster; nếu giá trị này đột ngột giảm và giữ nguyên, có thể suy ra có Job đã thất bại
• Có thể dùng taskmanager.cpu.load và taskmanager.memory.used để nắm mức sử dụng tài nguyên của cluster
• busyTimeMsPerSecond cho phép kiểm tra mức độ bận của TaskManager theo từng Job
• Nếu dùng Kafka làm source, có thể nhanh chóng kiểm tra độ trễ dữ liệu bằng records-lag-max

Ví dụ: tổng hợp window cho sự kiện đăng nhập Kafka

• Ví dụ trong Appendix nhận sự kiện từ Kafka và phát hành lên Kafka số lượng sự kiện đăng nhập trong 1 phút trước đó, cứ mỗi 10 giây
• Dữ liệu đầu vào ở định dạng JSON và gồm các trường event_time, event_type, data.user_id
• Truy vấn thiết lập pipeline.name, parallelism.default, table.exec.state.ttl
• Bảng đầu vào login_event dùng Kafka connector và JSON format, đồng thời đặt json.ignore-parse-errors thành true
• row_time được tạo từ event_time, và watermark được cấu hình để xử lý các sự kiện đến muộn tối đa 5 giây so với event time lớn nhất đã quan sát đến thời điểm hiện tại
• Bảng đầu ra windowed_login_count phát hành kết quả ra Kafka topic và bao gồm trường proc_time AS PROCTIME()
• HOP(row_time, INTERVAL '10' SECOND, INTERVAL '1' MINUTE) tạo hopping window 1 phút với bước 10 giây, và dùng COUNT(*) để tổng hợp số bản ghi trong window

Kết quả vận hành và các điểm còn cần cải thiện

• Nhóm đã tận dụng kinh nghiệm Flink sẵn có để bổ sung nhiều chức năng dễ hơn và nhanh hơn trước đây
• Nhóm đánh giá đã đạt được kết quả đáng hài lòng về năng suất và hiệu quả vận hành
• Sau khi áp dụng, hệ thống vận hành ổn định gần 1 năm mà hầu như không cần thêm công việc vận hành đáng kể
• Hiện tại đang được mở rộng dần trong vận hành thực tế
• Việc triển khai lại truy vấn và thay đổi cấu hình cluster vẫn còn bất tiện, và nhóm có kế hoạch cải thiện môi trường triển khai truy vấn bằng cách hiện thực mẫu GitOps Controller

[flgkselql98]

Các hệ thống phân tán như Flink cần duy trì HA bằng cách giữ 2~3 rack, và có vẻ như ở đây họ đã đảm bảo HA bằng cách tích hợp với Kubernetes. Nhưng rồi cuối cùng vẫn phải tính đến tài nguyên của các node worker của Kubernetes, nên tôi cũng tự hỏi liệu họ có cấu hình các node chỉ để chạy Flink hay không (có vẻ sẽ có vấn đề node worker bị sập khi tải Flink tăng cao).
Vậy từ góc nhìn đó, dùng Kubernetes có lợi thế gì không?

Ngoài ra, khi dùng window function trong Flink thì dữ liệu trong khoảng đó sẽ được giữ trong bộ nhớ để câu lệnh SQL join hoạt động, nên xét ở góc độ trade-off thì tôi cũng băn khoăn liệu Flink có phải là lựa chọn tốt hay không. Nếu theo thời gian, SQL ngày càng phình to mà job bị chết thì hậu quả sẽ rất lớn..

Bản thân tôi cũng đang suy nghĩ rằng khi cần join ở data source tầng trên cùng thì thay vì dùng Flink, có thể hạ xuống mức application level để xử lý theo cách nào.