Kisah Adopsi Flink SQL

 (hyperconnect.github.io)
3 poin oleh GN⁺ 2025-02-25 | 1 komentar | Bagikan ke WhatsApp
  • Hyperconnect Azar Matching Dev Team memilih menerapkan pemrosesan streaming dengan SQL, bukan kode aplikasi, untuk memecah aplikasi legacy Flink monolitik yang menggunakan 96 CPU
  • Jika dipecah menjadi beberapa Flink App, isolasi memang membaik tetapi beban operasional meningkat, sehingga tim menilai Flink SQL yang dapat memanfaatkan pemahaman internal tentang Flink milik tim lebih cocok dari sisi produktivitas dan efisiensi operasional
  • Flink SQL mendukung Checkpoint/Savepoint, JobManager HA, redistribusi TaskManager, window·join·event time·watermark, UDF, dan custom connector, sehingga lebih sesuai dengan kondisi tim dibanding ksqlDB dan Spark Structured Streaming
  • Lingkungan operasional disusun sebagai Session mode Flink Cluster di atas Kubernetes, dan deployment query serta penghentian Job ditangani dengan pendekatan GitOps menggunakan Flink SQL Gateway API dan GitHub Actions
  • Selama sekitar 1 tahun berjalan stabil dan sedang diperluas, tetapi redeployment query dan perubahan konfigurasi cluster masih merepotkan sehingga tim merencanakan perbaikan berbasis pola GitOps Controller

Latar belakang mengganti aplikasi streaming legacy yang berat

  • Azar Matching Dev Team mengelola beberapa aplikasi berbasis Flink, dan salah satunya adalah aplikasi legacy berat yang menggunakan 96 CPU
  • Aplikasi ini memiliki struktur monolitik yang menggabungkan berbagai fungsi di satu tempat, seperti join beberapa event matching, penerbitan event bersyarat, dan penyimpanan flag Redis
  • Setelah node eksekusi diubah karena pekerjaan infrastruktur perusahaan, aplikasi tidak berjalan normal, dan sulit menyelesaikannya dengan cepat hanya melalui tuning sederhana
  • Karena fungsi penting untuk join event sudah diimplementasikan di Flink app baru pada proyek terpisah, dibutuhkan cara untuk mengganti bagian setelah join event, yaitu penerbitan event bersyarat dan eksekusi logika

Perbandingan opsi pengganti

  • Jika diimplementasikan sebagai satu Flink App, objek yang dikelola memang lebih sedikit, tetapi kemungkinan besar akan kembali menjadi aplikasi besar, dan kegagalan di satu bagian dapat memengaruhi fungsi lain
  • Jika dipecah menjadi beberapa Flink App, setiap aplikasi dapat dikelola secara independen, tetapi semakin banyak aplikasi, semakin besar pula beban cluster, resource, dan deployment
  • Flink SQL memungkinkan pengembangan cepat dengan mendefinisikan logika melalui query dan cukup mengelola satu cluster, tetapi logika kompleks sulit diekspresikan hanya dengan SQL dan membutuhkan pengalaman operasional cluster
  • Tim sudah memiliki pemahaman yang lebih tinggi tentang implementasi internal Flink, dan menilai Flink SQL memiliki keunggulan dalam produktivitas dan efisiensi operasional

Alasan memilih Flink SQL

  • Flink SQL memungkinkan implementasi aplikasi pemrosesan streaming event dengan SQL tanpa harus menulis kode aplikasi secara langsung
  • Dari sisi high availability (HA), Flink mendukung pemrosesan stateful, dan dengan Checkpoint serta Savepoint, status pekerjaan dapat disimpan dan dipulihkan secara berkala atau pada titik waktu yang diinginkan
    • JobManager dapat dikonfigurasi dalam mode HA berbentuk leader-standby
    • Jika sebagian TaskManager gagal, pekerjaan TaskManager yang gagal dapat didistribusikan ulang ke TaskManager lain sesuai strategi retry Job
  • Fitur utama pemrosesan streaming dapat ditangani hanya dengan sintaks SQL
    • Mengubah bentuk data dengan SELECT dan memfilter record dengan WHERE
    • Menggabungkan beberapa stream dengan JOIN dan menyatukan stream dengan UNION
    • Mendukung pemrosesan window seperti tumbling, hopping(sliding), dan session window
    • Dapat mengatur toleransi data terlambat dengan pemrosesan event time dan watermark
  • Kebutuhan di luar fitur bawaan juga diperluas dengan UDF dan custom connector
    • Sebagian besar legacy yang ada menggunakan pola command Redis SET atau INCR, dan karena tidak ada Redis Connector resmi dari Flink, tim menulis sendiri Redis Connector dan menggunakannya
    • Saat itu belum ada fungsi built-in untuk mencari irisan tipe ARRAY, sehingga diimplementasikan sebagai UDF dan digunakan dalam query

Perbandingan dengan ksqlDB dan Spark Structured Streaming

  • ksqlDB termasuk dalam platform Confluent yang digunakan secara internal untuk Kafka, dan sudah ada use case di tingkat perusahaan
  • Namun, tim menilai ada inefisiensi pada perilaku HA untuk pemrosesan streaming stateful
    • Saat failover operasi stateful, semua changelog yang merupakan catatan perubahan state harus di-replay, sehingga waktu failover bisa lama
    • Pendekatan yang menempatkan replika stream pemrosesan dan terus memperbarui changelog ke state internal juga menjalankan operasi yang sama pada replika, sehingga resource dapat terpakai dua kali lipat
    • Informasi terkait dapat dilihat di Configuring ksqlDB for High Availability | Confluent Developer
  • Spark Structured Streaming adalah engine pemrosesan streaming berbasis engine Spark SQL
    • Ada use case internal, serta memungkinkan penulisan UDF dan Custom Sink
    • Memiliki ekosistem yang lebih besar dan lebih matang dibanding Flink
  • Spark berjalan dalam unit micro-batch sehingga dapat menimbulkan latensi per record, dan dalam situasi yang membutuhkan pemrosesan real-time, bisa kurang menguntungkan dibanding Flink
  • Karena pengalaman Spark di dalam tim hampir tidak ada dan penulisan Custom Sink juga diperlukan, tim sulit memilih Spark dengan yakin

Membangun lingkungan cluster

  • Di lokal, binary dapat diunduh dari halaman web resmi Flink dan cluster dapat dijalankan dengan {FLINK_HOME}/bin/start-cluster.sh
  • Jika menjalankan {FLINK_HOME}/bin/sql-client.sh, Flink SQL CLI akan terbuka, dan query uji seperti SELECT 1; dapat dikirimkan
  • Setelah query dikirimkan, di Flink web UI dapat dilihat bahwa query yang dikirim telah dikonversi menjadi Job dan dijalankan
  • Pada akhir 2022, Flink SQL Gateway dirilis sehingga pengiriman query berbasis HTTP menjadi memungkinkan

Arsitektur operasional berbasis Kubernetes

  • Karena sebagian besar layanan internal berjalan di atas Kubernetes, Flink SQL Cluster juga disusun di atas Kubernetes
  • Flink App yang sudah ada semuanya dideploy dan dioperasikan dalam Application mode
    • Ini adalah pendekatan menjalankan cluster terpisah untuk setiap aplikasi
    • Di Kubernetes, setiap aplikasi berjalan dengan masing-masing JobManager Pod dan TaskManager Pod
    • Pendekatan ini menguntungkan dari sisi independensi dan isolasi antar-aplikasi, serta pengelolaan konfigurasi dan dependensi per pekerjaan
  • Karena Flink SQL mengirimkan Job ke cluster yang sudah berjalan, JobManager dan TaskManager harus dijalankan dalam Session mode
  • Cluster disusun berdasarkan panduan Stand Alone Cluster on Kubernetes
  • Untuk lingkungan HA, tim merujuk konfigurasi High-Availability with Standalone Kubernetes, dan menggunakan s3 untuk high-availability.storageDir
  • Pendekatan Native Kubernetes dianggap tidak cocok dengan lingkungan infrastruktur internal yang mendefinisikan dan mendeploy konfigurasi deployment secara langsung, karena cluster dijalankan melalui shell script yang disediakan

Konfigurasi integrasi HA dan S3

  • Untuk integrasi HA dan S3, konfigurasi berikut digunakan di 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}
  • Dalam lingkungan HA, dua JobManager pod dijalankan, dan alamat keduanya harus berbeda agar logika pemilihan leader dan lainnya berjalan normal
  • Argumen eksekusi container JobManager diatur seperti berikut
args: ["start-foreground", "-D", "jobmanager.rpc.address=$(POD_IP)"]
  • Melalui konfigurasi ini, informasi JobManager pod yang saat ini terpilih sebagai leader dan Job ID yang sedang berjalan disimpan di Kubernetes ConfigMap dan digunakan untuk HA

Deployment query dengan pendekatan GitOps

  • Flink belum menyediakan web UI atau tool khusus untuk Flink SQL secara langsung
  • Kasus integrasi Hue ditinjau sebagai PoC, tetapi saat itu diperlukan pengembangan tambahan karena isu kompatibilitas versi Flink SQL Gateway, dan penyusunan lingkungan pengembangan juga memakan banyak waktu
  • Karena pola GitOps banyak digunakan secara internal, tim mengimplementasikan GitHub Actions untuk mendeploy query atau menghentikan Job
  • Di dalam repository, dibuat folder per Job dan query yang akan dijalankan dikumpulkan sebagai file SQL
  • GitHub Actions menerima nama folder dan menentukan file SQL tempat query akan diekstrak
  • Implementasinya berupa pemanggilan Flink SQL Gateway REST API, dan ditulis dengan Python yang sederhana serta mudah diuji

Contoh operasional dan penanganan insiden

  • Belum pernah ada pengalaman JobManager gagal, tetapi berdasarkan konfigurasi HA, jika JobManager gagal, JobManager lain dapat terpilih sebagai leader dan melanjutkan pekerjaan
  • TaskManager sesekali gagal, dan sebagian besar kasusnya adalah Pod direstart karena kebijakan Kubernetes QoS
  • Tim mengonfirmasi bahwa meskipun sebagian TaskManager gagal, pekerjaan didistribusikan ulang ke TaskManager lain dan tetap berlanjut
  • Kegagalan query sebagian besar terjadi karena masuknya data abnormal atau kekurangan resource komputasi
    • Saat membaca data JSON, format JSON yang salah dapat diabaikan sebagai data error dengan opsi json.ignore-parse-errors
    • Error yang terjadi saat mengekstrak data pada path tertentu dengan JSON_VALUE karena nilainya tidak ada atau tipenya berbeda dapat diberi nilai default dengan DEFAULT {VALUE} ON ERROR
    • Jika CPU TaskManager melebihi 100% atau memori tidak mencukupi, resource TaskManager ditingkatkan atau parallelism query dinaikkan lalu dideploy ulang
  • Saat cluster direstart karena perubahan konfigurasi cluster atau penambahan UDF, ada kasus sebagian Job gagal
    • Penyebabnya sering kali karena pengaturan Job timeout atau retry tidak tepat
    • Pengaturan timeout dan retry diubah agar Job tidak menghentikan retry terlalu cepat dan terus mencoba ulang sampai stabil setelah cluster restart

Perubahan query dan batasan pemulihan state

  • Saat kondisi query diubah lalu dideploy ulang, kasus di mana state dapat dipulihkan dengan savepoint terbatas pada perubahan yang sangat sederhana seperti perubahan nilai ekspresi kondisi
  • Jika kondisi window berubah, state juga berubah sehingga sulit menjaga kompatibilitas, dan pemulihan menggunakan savepoint bisa menjadi sulit
  • Jika state harus dipertahankan sementara kebutuhan sering berubah, menulis aplikasi secara langsung bisa menjadi pendekatan yang lebih baik

Poin monitoring

  • Flink menyediakan banyak metric bawaan, sehingga jika ada infrastruktur monitoring internal dan Metric Reporter yang sesuai, lingkungan monitoring dapat disusun dengan mudah
  • numRunningJobs menunjukkan jumlah Job yang sedang berjalan di cluster, dan jika nilainya tiba-tiba turun lalu bertahan, dapat dinilai ada Job yang gagal
  • taskmanager.cpu.load dan taskmanager.memory.used dapat digunakan untuk memahami penggunaan resource cluster
  • busyTimeMsPerSecond memungkinkan pengecekan seberapa sibuk TaskManager per Job
  • Jika menggunakan Kafka sebagai source, records-lag-max dapat digunakan untuk cepat memeriksa kondisi keterlambatan data

Contoh: agregasi window event login Kafka

  • Contoh Appendix menerima event dari Kafka dan menerbitkan jumlah event login selama 1 menit terakhir setiap 10 detik ke Kafka
  • Data input berformat JSON dan mencakup field event_time, event_type, dan data.user_id
  • Query mengatur pipeline.name, parallelism.default, dan table.exec.state.ttl
  • Tabel input login_event menggunakan Kafka connector dan JSON format, serta mengatur json.ignore-parse-errors menjadi true
  • row_time dibuat dari event_time, dan watermark diatur agar event yang datang terlambat hingga maksimal 5 detik dari event time terbesar yang sudah diamati tetap diproses
  • Tabel output windowed_login_count menerbitkan hasil ke topic Kafka, dan mencakup field proc_time AS PROCTIME()
  • Dengan HOP(row_time, INTERVAL '10' SECOND, INTERVAL '1' MINUTE), dibuat hopping window 1 menit dengan interval 10 detik, lalu jumlah record dalam window diagregasi dengan COUNT(*)

Hasil operasional dan perbaikan yang tersisa

  • Tim dapat memanfaatkan pengalaman Flink yang sudah ada untuk menambahkan berbagai fungsi dengan lebih mudah dan cepat dibanding sebelumnya
  • Tim menilai hasilnya memuaskan dari sisi produktivitas dan efisiensi operasional
  • Setelah adopsi, sistem berjalan stabil selama sekitar 1 tahun tanpa pekerjaan operasional berarti
  • Saat ini operasionalnya sedang diperluas secara bertahap
  • Masih ada ketidaknyamanan dalam redeployment query dan perubahan konfigurasi cluster, dan tim berencana memperbaiki lingkungan deployment query melalui implementasi pola GitOps Controller

Bacaan terkait

1 komentar

 
flgkselql98 2025-02-26

Sistem terdistribusi seperti Flink perlu mempertahankan 2–3 rack untuk menjaga HA, dan tampaknya dengan mengintegrasikan Kubernetes, HA berhasil dijamin. Namun pada akhirnya tetap perlu memikirkan resource untuk kube slave node, jadi saya jadi bertanya-tanya apakah mereka membangun node yang hanya menjalankan Flink saja (sepertinya bisa ada isu slave node down saat beban Flink tinggi).
Dari sudut pandang itu, apakah ada keuntungan menggunakan Kubernetes?

Selain itu, ketika memakai window function di Flink, data di antaranya akan dipertahankan di memori sehingga pernyataan SQL join bisa berjalan. Jika dilihat dari sudut pandang trade-off, apakah Flink benar-benar pilihan yang baik? Jika seiring waktu SQL + job makin besar lalu job-nya mati, akibatnya akan sangat besar..

Saya juga sedang memikirkan, saat join diperlukan di data source paling atas, alih-alih memakai Flink, dengan cara seperti apa pemrosesannya bisa diturunkan ke level application.