Flink SQL’i Benimseme Hikayesi

 (hyperconnect.github.io)
3 puan yazan GN⁺ 2025-02-25 | 1 yorum | WhatsApp'ta paylaş
  • Hyperconnect Azar Matching Dev Team, 96 CPU kullanan monolitik Flink legacy uygulamasını parçalamak için, akış işlemeyi uygulama kodu yerine SQL ile gerçekleştirme yöntemini seçti
  • Birden fazla Flink App’e bölmek izolasyonu iyileştirse de operasyon yükünü artırdığı için, ekibin Flink’in iç işleyişine dair bilgisinden yararlanabilen Flink SQL’in üretkenlik ve operasyonel verimlilik açısından daha uygun olduğuna karar verildi
  • Flink SQL; Checkpoint/Savepoint, JobManager HA, TaskManager yeniden dağıtımı, window/join/event time/watermark, UDF ve özel connector desteği sunduğu için ekibin koşullarına ksqlDB ve Spark Structured Streaming’den daha uygundu
  • Operasyon ortamı Kubernetes üzerinde Session mode Flink Cluster olarak kurgulandı; Flink SQL Gateway API ve GitHub Actions kullanılarak GitOps yaklaşımıyla sorgu dağıtımı ve Job durdurma işlemleri yönetildi
  • Yaklaşık 1 yıldır kararlı şekilde çalışıyor ve kapsamı genişletiliyor; ancak sorgu yeniden dağıtımı ve cluster yapılandırma değişiklikleri hâlâ zahmetli olduğundan GitOps Controller pattern tabanlı iyileştirmeler planlanıyor

Ağır legacy streaming uygulamasını değiştirme arka planı

  • Azar Matching Dev Team, Flink tabanlı birden fazla uygulamayı yönetiyordu; bunlardan biri 96 CPU kullanan ağır bir legacy uygulamaydı
  • Bu uygulama, birden fazla eşleşme olayını join etme, koşullu event yayınlama ve Redis flag saklama gibi işlevleri tek yerde toplayan monolitik bir yapıdaydı
  • Şirket genelindeki altyapı çalışmasıyla çalıştırma node’ları değiştirildikten sonra uygulama düzgün çalışmadı ve yalnızca basit tuning ile hızlıca çözülmesi zordu
  • Kritik event join işlevi ayrı bir projenin yeni Flink uygulamasında zaten uygulanmış olduğundan, event join sonrasındaki koşullu event yayınlama ve logic çalıştırma kısmını değiştirecek bir yöntem gerekiyordu

Alternatif yöntemlerin karşılaştırması

  • Tek bir Flink App olarak uygulanırsa yönetilecek hedef sayısı az olur; ancak yeniden devasa bir uygulamaya dönüşme olasılığı yüksektir ve bir bölümdeki hata diğer işlevleri etkileyebilir
  • Birden fazla Flink App’e bölünürse her uygulama bağımsız yönetilebilir; ancak uygulama sayısı arttıkça cluster, kaynak ve dağıtım yükü de artar
  • Flink SQL, logic’i sorgularla tanımlayarak hızlı geliştirme yapmayı ve yalnızca tek bir cluster yönetmeyi sağlar; ancak karmaşık logic’i yalnızca SQL ile ifade etmek zordur ve cluster operasyon deneyimi gerektirir
  • Ekip, Flink’in iç implementasyonuna dair anlayışının arttığı bir noktadaydı ve Flink SQL’in üretkenlik ve operasyonel verimlilik açısından avantaj sağladığına karar verdi

Flink SQL’in seçilme nedeni

  • Flink SQL, uygulama kodunu doğrudan yazmadan SQL ile event streaming processing app uygulanmasını sağlar
  • Yüksek erişilebilirlik (HA) açısından Flink stateful processing’i destekler; Checkpoint ve Savepoint ile job state’i periyodik olarak veya istenen bir anda kaydedilip geri yüklenebilir
    • JobManager, leader-standby biçiminde HA modunda yapılandırılabilir
    • TaskManager’ların bir kısmı başarısız olursa Job retry stratejisine göre başarısız TaskManager’ın işleri diğer TaskManager’lara yeniden dağıtılabilir
  • Yalnızca SQL söz dizimiyle de başlıca streaming processing işlevleri ele alınabilir
    • SELECT ile veri biçimi dönüştürülür, WHERE ile record’lar filtrelenir
    • JOIN ile birden fazla stream birleştirilir, UNION ile stream’ler bir araya getirilebilir
    • tumbling, hopping(sliding), session window gibi window processing desteklenir
    • event time processing ve watermark ile geciken veriler için kabul aralığı ayarlanabilir
  • UDF ve özel connector ile temel işlevlerin dışındaki gereksinimler de genişletilir
    • Mevcut legacy’nin büyük kısmı Redis SET veya INCR komutlarını kullanan pattern’lerden oluşuyordu; Flink’in resmi Redis Connector’ı olmadığından Redis Connector doğrudan yazılıp kullanıldı
    • O dönemde ARRAY tipinin kesişimini alan built-in bir fonksiyon olmadığından UDF olarak uygulanıp sorgularda kullanıldı

ksqlDB ve Spark Structured Streaming ile karşılaştırma

  • ksqlDB, şirket içinde Kafka için kullanılan Confluent platformuna dahildi ve şirket genelinde kullanım örnekleri de vardı
  • Ancak stateful streaming processing’in HA davranışında verimsizlikler olduğu değerlendirildi
    • Stateful operation failover sırasında state değişiklik kaydı olan changelog’un tamamının replay edilmesi gerektiğinden failover süresi uzayabilir
    • Processing stream replica’ları tutup changelog’u internal state’e sürekli update eden yöntem, replica’larda da aynı operation’ı çalıştırdığı için kaynak tüketimini iki katına çıkarabilir
    • İlgili içerik Configuring ksqlDB for High Availability | Confluent Developer adresinden incelenebilir
  • Spark Structured Streaming, Spark SQL engine tabanlı bir streaming processing engine’dir
    • Şirket içinde kullanım örnekleri vardır; UDF ve Custom Sink yazımı mümkündür
    • Flink’e kıyasla daha büyük ve iyi kurulmuş bir ekosisteme sahiptir
  • Spark, micro-batch birimleriyle çalıştığı için record bazında latency oluşabilir; gerçek zamanlı işlemenin önemli olduğu durumlarda Flink’e göre dezavantajlı olabilir
  • Ekip içinde Spark deneyimi neredeyse yoktu ve Custom Sink yazımı da gerektiğinden Spark’ı rahatça seçmek zordu

Cluster ortamının kurulumu

  • Lokal ortamda Flink resmi web sayfasından binary indirilip {FLINK_HOME}/bin/start-cluster.sh ile cluster çalıştırılabilir
  • {FLINK_HOME}/bin/sql-client.sh çalıştırıldığında Flink SQL CLI açılır ve SELECT 1; gibi test sorguları gönderilebilir
  • Sorgu gönderildikten sonra Flink web UI’da gönderilen sorgunun Job’a dönüştürülüp çalıştırıldığı görülebilir
  • 2022 sonunda Flink SQL Gateway yayınlanınca HTTP tabanlı sorgu gönderimi mümkün hale geldi

Kubernetes tabanlı operasyon mimarisi

  • Şirket içindeki servislerin çoğu Kubernetes üzerinde çalıştığından Flink SQL Cluster da Kubernetes üzerinde yapılandırıldı
  • Mevcut Flink App’lerin tamamı Application mode ile dağıtılıp işletiliyordu
    • Her uygulama için ayrı bir cluster ayağa kaldırılan yöntemdir
    • Kubernetes’te her app için ayrı JobManager Pod ve TaskManager Pod çalıştırılır
    • Uygulamalar arası bağımsızlık ve izolasyon, iş bazlı ayarlar ve dependency yönetimi açısından avantajlıydı
  • Flink SQL, zaten çalışmakta olan bir cluster’a Job gönderdiği için JobManager ve TaskManager’ın Session mode ile çalıştırılması gerekiyordu
  • Cluster, Stand Alone Cluster on Kubernetes kılavuzu temel alınarak yapılandırıldı
  • HA ortamı için High-Availability with Standalone Kubernetes ayarları referans alındı; high-availability.storageDir için s3 kullanıldı
  • Native Kubernetes yöntemi, cluster’ı sağlanan shell script ile ayağa kaldırdığı için deployment ayarlarını doğrudan tanımlayıp dağıtan şirket içi altyapı ortamına uygun olmadığı değerlendirildi

HA ve S3 entegrasyon ayarları

  • HA ve S3 entegrasyonu için config.yaml içinde aşağıdaki ayarlar kullanıldı
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}
  • HA ortamında iki JobManager pod çalıştırılır; leader election logic’inin vb. düzgün çalışması için birbirlerinin adreslerinin farklı olması gerekir
  • JobManager container çalıştırma argümanları aşağıdaki gibi ayarlanır
args: ["start-foreground", "-D", "jobmanager.rpc.address=$(POD_IP)"]
  • Bu ayar sayesinde Kubernetes ConfigMap’e, mevcut leader olarak seçilen JobManager pod bilgisi ve o anda çalışan Job ID’leri gibi bilgiler kaydedilir ve HA için kullanılır

GitOps yaklaşımıyla sorgu dağıtımı

  • Flink tarafından doğrudan sunulan Flink SQL’e özel bir web UI veya dedicated tool henüz yok
  • Hue entegrasyon örneği PoC olarak incelendi; ancak o dönemde Flink SQL Gateway sürüm uyumluluğu sorunları nedeniyle ek geliştirme gerekiyordu ve geliştirme ortamını kurmak da çok zaman alıyordu
  • Şirket içinde GitOps pattern’i yaygın kullanıldığı için sorgu dağıtmak veya Job durdurmak üzere GitHub Actions uygulandı
  • Repository içinde Job bazlı klasörler oluşturuldu ve çalıştırılacak sorgular SQL dosyaları olarak toplandı
  • GitHub Actions, klasör adını alarak sorguların çıkarılacağı SQL dosyasını belirler
  • Implementasyon, Flink SQL Gateway REST API çağırma şeklindedir; basit ve test etmesi kolay olduğu için Python ile yazıldı

Operasyon örnekleri ve arıza müdahalesi

  • JobManager’ın başarısız olduğu bir deneyim yaşanmadı; ancak HA ayarlarına göre JobManager başarısız olursa başka bir JobManager leader seçilerek işi sürdürebilir
  • TaskManager zaman zaman başarısız oldu; çoğu Kubernetes QoS policy nedeniyle Pod’un yeniden başlatıldığı vakalardı
  • Bazı TaskManager’lar başarısız olsa bile işlerin diğer TaskManager’lara yeniden dağıtılıp devam ettiği doğrulandı
  • Sorgu hatalarının çoğu anormal veri girişi veya compute kaynağı yetersizliği nedeniyle oluşur
    • JSON verisi okunurken hatalı JSON formatı, json.ignore-parse-errors seçeneğiyle hata veren veriler yok sayılarak yönetilebilir
    • JSON_VALUE ile belirli bir path’teki veri çıkarılırken değer yoksa veya tip farklıysa oluşan hatalar için DEFAULT {VALUE} ON ERROR ile varsayılan değer ayarlanabilir
    • TaskManager CPU’su %100’ü aşarsa veya bellek yetersizse TaskManager kaynakları artırılır ya da sorgu parallelism’i artırılıp yeniden dağıtım yapılır
  • Cluster ayarı değişikliği veya UDF ekleme nedeniyle cluster yeniden başlatıldığında bazı Job’ların başarısız olduğu durumlar yaşandı
    • Neden çoğunlukla Job timeout veya retry ayarlarının uygun olmamasıydı
    • Job’un retry’ı çok hızlı bitirmemesi ve cluster yeniden başlatıldıktan sonra stabil hale gelene kadar yeniden denemesi için timeout ve retry ayarları değiştirildi

Sorgu değişikliği ve state geri yükleme kısıtları

  • Sorgu koşulları değiştirilip yeniden dağıtıldığında savepoint ile state’in geri yüklenebildiği durumlar, koşul ifadesi değerinin değişmesi gibi çok basit değişikliklerle sınırlıdır
  • Window koşulu değişirse state de değiştiğinden uyumluluğu korumak zorlaşır ve savepoint kullanarak geri yükleme zor olabilir
  • State’in korunması gerekiyorsa ancak gereksinimler sık değişiyorsa doğrudan app yazma yöntemi daha iyi olabilir

İzleme noktaları

  • Flink birçok varsayılan metric sunduğundan, şirket içi monitoring altyapısı ve uygun bir Metric Reporter varsa monitoring ortamı kolayca kurulabilir
  • numRunningJobs, cluster’da o anda çalışan Job sayısını gösterir; değer aniden düşüp düşük kalırsa başarısız bir Job olduğu düşünülebilir
  • taskmanager.cpu.load ve taskmanager.memory.used ile cluster kaynak kullanımı anlaşılabilir
  • busyTimeMsPerSecond ile TaskManager’ın ne kadar meşgul olduğu Job bazında görülebilir
  • Kafka source olarak kullanılıyorsa records-lag-max ile veri gecikme durumu hızlıca kontrol edilebilir

Örnek: Kafka login event window aggregation

  • Appendix örneği, Kafka’dan event alıp her 10 saniyede bir son 1 dakikadaki login event sayısını Kafka’ya yayınlar
  • Girdi verisi JSON formatındadır ve event_time, event_type, data.user_id alanlarını içerir
  • Sorgu pipeline.name, parallelism.default, table.exec.state.ttl ayarlarını yapar
  • Girdi tablosu login_event, Kafka connector ve JSON format kullanır; json.ignore-parse-errors değeri true olarak ayarlanır
  • row_time, event_time üzerinden oluşturulur; watermark, şimdiye kadar gözlemlenen event time’a göre en fazla 5 saniye geç gelen event’leri işleyecek şekilde ayarlanır
  • Çıktı tablosu windowed_login_count, sonucu Kafka topic’e yayınlar ve proc_time AS PROCTIME() alanını içerir
  • HOP(row_time, INTERVAL '10' SECOND, INTERVAL '1' MINUTE) ile 10 saniye aralıklı 1 dakikalık hopping window oluşturulur ve COUNT(*) ile window içindeki record sayısı aggregate edilir

Operasyon sonucu ve kalan iyileştirmeler

  • Ekip, mevcut Flink deneyiminden yararlanarak önceye kıyasla birden fazla işlevi daha kolay ve hızlı ekleyebildi
  • Üretkenlik ve operasyonel verimlilik açısından tatmin edici sonuçlar elde edildiğini değerlendirdi
  • Benimsenmesinden sonra yaklaşık 1 yıl boyunca belirgin bir operasyon işi olmadan kararlı şekilde çalıştı
  • Şu anda kapsamı kademeli olarak genişletiliyor
  • Sorgu yeniden dağıtımı ve cluster ayar değişikliklerinde hâlâ rahatsızlıklar var; GitOps Controller pattern implementasyonu ile sorgu dağıtım ortamının iyileştirilmesi planlanıyor

İlgili okumalar

1 yorum

 
flgkselql98 2025-02-26

Flink gibi dağıtık sistemlerde HA’yi korumak için 2~3 rack bulundurmak gerekir; ancak Kubernetes entegrasyonuyla bunun sağlandığı anlaşılıyor. Yine de sonuçta kube slave node tarafındaki kaynakları da düşünmek gerekecek; acaba sadece Flink çalıştıran node’lardan oluşan bir yapı mı kurdular diye düşündürüyor (Flink yükü sırasında slave node’ların düşmesi gibi bir sorun olabilir).
Bu açıdan bakınca Kubernetes kullanmanın bir avantajı var mı?

Ayrıca Flink’te window function kullanıldığında, o aralıktaki veri bellekte tutulurken SQL join ifadesi çalışıyor; bu trade-off açısından bakıldığında Flink gerçekten iyi bir tercih mi diye düşünüyorum. Zaman geçtikçe büyüyen SQL + job öldüğünde ortaya çıkacak büyük sorunlar...

Ben de en üst seviyedeki data source tarafında join gereken durumlarda Flink kullanmadan, bunu application level’a indirip hangi yöntemle işleyebileceğimizi düşünüyorum.