Flink SQL अपनाने की कहानी

 (hyperconnect.github.io)
3 पॉइंट द्वारा GN⁺ 2025-02-25 | 1 टिप्पणियां | WhatsApp पर शेयर करें
  • Hyperconnect Azar Matching Dev Team ने CPU 96 cores इस्तेमाल करने वाले monolithic Flink legacy app को तोड़ने के लिए, application code के बजाय SQL से streaming processing लागू करने का तरीका चुना
  • कई Flink Apps में बाँटने से isolation बेहतर होता, लेकिन operations का बोझ बढ़ जाता; इसलिए टीम की Flink internals की समझ का लाभ उठाने वाला Flink SQL productivity और operational efficiency के लिहाज से ज़्यादा उपयुक्त लगा
  • Flink SQL Checkpoint/Savepoint, JobManager HA, TaskManager redistribution, windows·joins·event time·watermarks, UDF और custom connectors को support करता है, इसलिए ksqlDB और Spark Structured Streaming की तुलना में टीम की स्थिति के लिए बेहतर था
  • operating environment Kubernetes पर Session mode Flink Cluster के रूप में बनाया गया, और Flink SQL Gateway API व GitHub Actions का उपयोग कर GitOps तरीके से query deployment और Job termination संभाला गया
  • लगभग 1 साल से stable तरीके से चल रहा है और विस्तार हो रहा है, लेकिन query redeployment और cluster configuration changes अभी भी असुविधाजनक हैं, इसलिए GitOps Controller pattern आधारित सुधार की योजना है

भारी legacy streaming app को बदलने की पृष्ठभूमि

  • Azar Matching Dev Team कई Flink-based apps manage कर रही थी, और उनमें से एक भारी legacy app था जो 96 CPU cores इस्तेमाल करता था
  • यह app कई match event joins, conditional event publishing, Redis flags save करने जैसी functionalities को एक जगह इकट्ठा करने वाला monolithic structure था
  • company-wide infrastructure work के तहत execution nodes बदलने के बाद app सही तरह से काम नहीं कर रहा था, और केवल simple tuning से इसे जल्दी हल करना मुश्किल था
  • अहम event join functionality एक अलग project के नए Flink app में पहले ही implement हो चुकी थी, इसलिए event join के बाद conditional event publishing और logic execution वाले हिस्से को replace करने का तरीका चाहिए था

replacement approaches की तुलना

  • एक Flink App के रूप में implement करने पर manage करने वाली चीज़ें कम होती हैं, लेकिन फिर से एक बड़ा app बनने की संभावना ज़्यादा रहती है और एक हिस्से की failure दूसरी functionalities को प्रभावित कर सकती है
  • कई Flink Apps में बाँटने पर हर app को independently manage किया जा सकता है, लेकिन apps की संख्या बढ़ने के साथ cluster·resource·deployment का बोझ भी बढ़ता है
  • Flink SQL queries से logic define करके तेज़ development और केवल एक cluster manage करने की सुविधा देता है, लेकिन complex logic को केवल SQL में express करना कठिन है और cluster operations का अनुभव चाहिए
  • टीम की Flink internal implementation की understanding बेहतर हो चुकी थी, और उसने माना कि Flink SQL productivity और operational efficiency में फायदे देता है

Flink SQL चुनने के कारण

  • Flink SQL application code सीधे लिखे बिना SQL से event streaming processing app implement कर सकता है
  • High Availability (HA) के संदर्भ में Flink stateful processing support करता है, और Checkpoint व Savepoint से job state को periodic या इच्छित point in time पर save·restore किया जा सकता है
    • JobManager को leader-standby form के HA mode में configure किया जा सकता है
    • यदि कुछ TaskManagers fail हो जाएँ, तो Job retry strategy के अनुसार failed TaskManager के tasks दूसरे TaskManager पर redistribute किए जा सकते हैं
  • केवल SQL syntax से भी मुख्य streaming processing features handle किए जा सकते हैं
    • SELECT से data shape transform किया जाता है और WHERE से records filter किए जाते हैं
    • JOIN से कई streams combine किए जाते हैं और UNION से streams merge किए जा सकते हैं
    • tumbling, hopping(sliding), session window जैसे window processing को support करता है
    • event time processing और watermarks से late data की allowed range set की जा सकती है
  • UDF और custom connectors से default functionality से बाहर की requirements भी extend की गईं
    • मौजूदा legacy का अधिकांश हिस्सा Redis SET या INCR commands इस्तेमाल करने वाला pattern था, और Flink official Redis Connector नहीं था, इसलिए Redis Connector खुद लिखकर इस्तेमाल किया गया
    • उस समय ARRAY type का intersection निकालने वाला built-in function नहीं था, इसलिए UDF से implement करके query में इस्तेमाल किया गया

ksqlDB·Spark Structured Streaming से तुलना

  • ksqlDB company में Kafka के लिए इस्तेमाल होने वाले Confluent platform में शामिल था, और company-wide use cases भी थे
  • लेकिन stateful streaming processing के HA behavior में inefficiency मानी गई
    • stateful operation failover के समय state changes का record यानी changelog पूरा replay करना पड़ता है, जिससे failover time लंबा हो सकता है
    • processing stream replica रखकर changelog को internal state में लगातार update करने वाला तरीका replica पर भी वही operation करता है, इसलिए resources दोगुने consume हो सकते हैं
    • संबंधित जानकारी Configuring ksqlDB for High Availability | Confluent Developer में देखी जा सकती है
  • Spark Structured Streaming Spark SQL engine आधारित streaming processing engine है
    • company में इसके use cases हैं और UDF व Custom Sink लिखना संभव है
    • Flink की तुलना में इसका ecosystem बड़ा और अच्छी तरह स्थापित है
  • Spark micro-batch units में काम करता है, इसलिए record-level latency हो सकती है, और real-time processing महत्वपूर्ण होने की स्थिति में यह Flink की तुलना में disadvantage में हो सकता है
  • टीम के भीतर Spark experience लगभग नहीं था और Custom Sink लिखने की भी ज़रूरत थी, इसलिए Spark को सहजता से चुनना कठिन था

cluster environment setup

  • local में Flink official webpage से binary download करके {FLINK_HOME}/bin/start-cluster.sh से cluster run किया जा सकता है
  • {FLINK_HOME}/bin/sql-client.sh run करने पर Flink SQL CLI खुलता है, और SELECT 1; जैसी test query submit की जा सकती है
  • query submit करने के बाद Flink web UI में देखा जा सकता है कि submitted query Job में convert होकर run हुई है
  • 2022 के अंत में Flink SQL Gateway release होने के बाद HTTP-based query submission संभव हुआ

Kubernetes-based operations architecture

  • company की अधिकांश services Kubernetes पर चलती हैं, इसलिए Flink SQL Cluster भी Kubernetes पर बनाया गया
  • existing Flink Apps सभी Application mode में deploy·operate हो रहे थे
    • हर application के लिए अलग cluster चलाने का तरीका है
    • Kubernetes में हर app के लिए JobManager Pod और TaskManager Pod अलग-अलग launch होकर चलते हैं
    • apps के बीच independence और isolation, task-specific configuration और dependency management के लिहाज से लाभदायक था
  • Flink SQL पहले से चल रहे cluster में Job submit करता है, इसलिए JobManager और TaskManager को Session mode में चलाना था
  • cluster Stand Alone Cluster on Kubernetes guide के आधार पर बनाया गया
  • HA environment के लिए High-Availability with Standalone Kubernetes settings को reference किया गया, और high-availability.storageDir के लिए s3 इस्तेमाल किया गया
  • Native Kubernetes तरीका provided shell script से cluster चलाता है, इसलिए deployment settings को सीधे define करके deploy करने वाले internal infra environment के लिए उपयुक्त नहीं माना गया

HA और S3 integration settings

  • HA और S3 integration के लिए config.yaml में नीचे की settings इस्तेमाल की गईं
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 environment में दो JobManager pods चलाए जाते हैं, और उनके addresses अलग होने चाहिए तभी leader election logic आदि सही तरह से काम करते हैं
  • JobManager container execution arguments नीचे की तरह set किए गए
args: ["start-foreground", "-D", "jobmanager.rpc.address=$(POD_IP)"]
  • इस setting के जरिए Kubernetes ConfigMap में current leader के रूप में चुने गए JobManager pod की जानकारी और वर्तमान में running Job IDs आदि save होकर HA में इस्तेमाल होते हैं

GitOps तरीके से query deployment

  • Flink सीधे Flink SQL के लिए web UI या dedicated tool अभी provide नहीं करता
  • Hue integration case को PoC के रूप में review किया गया, लेकिन उस समय Flink SQL Gateway version compatibility issue के कारण additional development चाहिए था और development environment setup में भी काफी समय लगता
  • company में GitOps pattern का बहुत उपयोग होता है, इसलिए query deploy करने या Job stop करने वाले GitHub Actions implement किए गए
  • Repository के भीतर Job-wise folders बनाए गए और execute होने वाली queries को SQL files में collect करके रखा गया
  • GitHub Actions folder name लेकर query extract करने वाली SQL file specify करता है
  • implementation Flink SQL Gateway REST API call करने के तरीके से है, और इसे simple व test करने में आसान Python में लिखा गया

operations cases और incident handling

  • JobManager fail होने का अनुभव नहीं था, लेकिन HA settings के अनुसार JobManager fail होने पर दूसरा JobManager leader के रूप में elected होकर work continue कर सकता है
  • TaskManager कभी-कभी fail हुआ, और अधिकांश cases Kubernetes QoS policy के कारण Pod restart होने के थे
  • पुष्टि हुई कि कुछ TaskManagers fail होने पर भी work दूसरे TaskManagers पर redistribute होकर चलता रहता है
  • query failures ज्यादातर abnormal data ingestion या computing resources की कमी के कारण हुईं
    • JSON data पढ़ते समय invalid JSON format को json.ignore-parse-errors option से error data ignore किया जा सकता है
    • JSON_VALUE से specific path का data extract करते समय value missing होने या type mismatch से होने वाले errors के लिए DEFAULT {VALUE} ON ERROR से default value set की जा सकती है
    • TaskManager CPU 100% से ऊपर जाने या memory कम होने पर TaskManager resources बढ़ाए गए या query parallelism बढ़ाकर redeploy किया गया
  • cluster settings change करने या UDF add करने के लिए cluster restart करने पर कुछ Jobs fail होने के cases थे
    • अक्सर कारण Job timeout या retry settings का appropriate न होना था
    • timeout और retry settings को modify किया गया ताकि Job बहुत जल्दी retry बंद न करे और cluster restart के बाद stable होने तक retry करता रहे

query changes और state restore की limitations

  • query conditions modify करके redeploy करते समय savepoint से state restore कर पाने के cases condition expression value change जैसे बहुत simple modifications तक सीमित हैं
  • window condition बदलने पर state भी बदल जाता है, जिससे compatibility maintain करना कठिन हो जाता है, और savepoint-based restore मुश्किल हो सकता है
  • यदि state maintain करना जरूरी है और requirements अक्सर बदलती हैं, तो सीधे app लिखने का तरीका बेहतर हो सकता है

monitoring points

  • Flink में बहुत सारे built-in metrics होते हैं, इसलिए internal monitoring infra और suitable Metric Reporter होने पर monitoring environment आसानी से बनाया जा सकता है
  • numRunningJobs cluster में currently running Jobs की संख्या दिखाता है; यदि value अचानक घटकर बनी रहे तो failed Job होने का अनुमान लगाया जा सकता है
  • taskmanager.cpu.load और taskmanager.memory.used से cluster resource usage समझा जा सकता है
  • busyTimeMsPerSecond से Job-wise देखा जा सकता है कि TaskManager कितना busy है
  • Kafka को source के रूप में इस्तेमाल करने पर records-lag-max से data lag status जल्दी देखा जा सकता है

example: Kafka login event window aggregation

  • Appendix example Kafka से events लेकर हर 10 सेकंड में पिछले 1 minute के login events की संख्या Kafka पर publish करता है
  • input data JSON format में है और event_time, event_type, data.user_id fields शामिल करता है
  • query pipeline.name, parallelism.default, table.exec.state.ttl set करती है
  • input table login_event Kafka connector और JSON format इस्तेमाल करता है, और json.ignore-parse-errors को true set करता है
  • row_time event_time से generate होता है, और watermark current time तक observe किए गए event time के आधार पर maximum 5 seconds late arriving events process करने के लिए set किया गया है
  • output table windowed_login_count Kafka topic पर results publish करता है, और proc_time AS PROCTIME() field शामिल करता है
  • HOP(row_time, INTERVAL '10' SECOND, INTERVAL '1' MINUTE) से 10-second interval वाला 1-minute hopping window बनाया जाता है, और COUNT(*) से window के भीतर records की संख्या aggregate की जाती है

operations results और बाकी improvements

  • टीम existing Flink experience का इस्तेमाल करके पहले की तुलना में आसानी और तेजी से कई features add कर सकी
  • productivity और operational efficiency के लिहाज से satisfactory results मिले, ऐसा मूल्यांकन किया गया
  • adoption के बाद लगभग 1 year तक बिना किसी खास operations work के stable तरीके से चलता रहा
  • वर्तमान में धीरे-धीरे operation का विस्तार किया जा रहा है
  • query redeployment और cluster configuration changes में अभी भी असुविधा बची है, और GitOps Controller pattern implement करके query deployment environment सुधारने की योजना है

संबंधित पढ़ाई

1 टिप्पणियां

 
flgkselql98 2025-02-26

ऐसा लगता है कि flink जैसे distributed systems में 2~3 rack बनाए रखकर HA बनाए रखना पड़ता है, लेकिन Kubernetes को जोड़कर HA सुनिश्चित किया गया है। लेकिन आखिरकार kube slave node के resources पर भी विचार करना होगा, तो सोचता हूँ कि क्या उन्होंने सिर्फ flink चलाने वाले nodes बनाए हैं (flink पर load बढ़ने पर slave node down होने की समस्या हो सकती है)।
उस नज़रिए से Kubernetes इस्तेमाल करने का फ़ायदा क्या होगा?

साथ ही, flink में window function इस्तेमाल करने पर उस दौरान data memory में बना रहता है और उसी से SQL join statement काम करता है। trade-off के नज़रिए से देखें तो क्या flink वाकई अच्छा विकल्प है, यह सोचने वाली बात है। समय के साथ बहुत बड़ा होता जाने वाला SQL + job अगर मर जाए तो उससे पैदा होने वाली भारी समस्या...

मैं भी सोच रहा हूँ कि जब सबसे ऊपर वाले data source पर join की ज़रूरत हो, तब flink का इस्तेमाल किए बिना किस तरह इसे application level पर नीचे लाकर process किया जा सकता है।