บันทึกการนำ Flink SQL มาใช้

 (hyperconnect.github.io)
3 คะแนน โดย GN⁺ 2025-02-25 | 1 ความคิดเห็น | แชร์ทาง WhatsApp
  • ทีม Hyperconnect Azar Matching Dev Team เลือกใช้วิธีเขียนการประมวลผลสตรีมมิงด้วย SQL แทนโค้ดแอปพลิเคชัน เพื่อแยก แอป Flink เลกาซีแบบโมโนลิธิก ที่ใช้ CPU 96 คอร์ออกเป็นส่วน ๆ
  • หากแยกเป็น Flink App หลายตัว การแยกส่วนจะดีขึ้นแต่ภาระด้านการปฏิบัติการจะเพิ่มขึ้น ทีมจึงมองว่า Flink SQL ซึ่งใช้ประโยชน์จาก ความเข้าใจภายในของ Flink ที่ทีมมีอยู่ เหมาะสมกว่าในแง่ผลิตภาพและประสิทธิภาพการปฏิบัติการ
  • Flink SQL รองรับ Checkpoint/Savepoint, JobManager HA, การกระจาย TaskManager ใหม่, window/join/event time/watermark, UDF และ custom connector จึงเหมาะกับสถานการณ์ของทีมมากกว่า ksqlDB และ Spark Structured Streaming
  • สภาพแวดล้อมปฏิบัติการถูกจัดเป็น Session mode Flink Cluster บน Kubernetes และใช้ Flink SQL Gateway API กับ GitHub Actions เพื่อจัดการการ deploy query และการหยุด Job ในรูปแบบ GitOps
  • หลังใช้งานอย่างเสถียรมาประมาณ 1 ปีและกำลังขยายการใช้งาน แต่การ redeploy query และการเปลี่ยน config ของ cluster ยังไม่สะดวก จึงวางแผนปรับปรุงโดยอิง GitOps Controller pattern

เบื้องหลังการแทนที่แอปสตรีมมิงเลกาซีขนาดใหญ่

  • Azar Matching Dev Team ดูแลแอปที่ใช้ Flink อยู่หลายตัว และในนั้นมีแอปเลกาซีขนาดใหญ่ที่ใช้ CPU 96 คอร์
  • แอปนี้เป็น โครงสร้างแบบโมโนลิธิก ที่รวมฟังก์ชันหลายอย่างไว้ในที่เดียว เช่น การ join อีเวนต์ matching หลายรายการ การส่งอีเวนต์แบบมีเงื่อนไข และการบันทึก flag ลง Redis
  • หลังจากเปลี่ยน node ที่ใช้รันจากงานด้านโครงสร้างพื้นฐานระดับบริษัท แอปก็ทำงานไม่ปกติ และยากจะแก้ได้อย่างรวดเร็วด้วยการปรับจูนเพียงอย่างเดียว
  • เนื่องจากฟังก์ชัน join อีเวนต์ที่สำคัญถูก implement ไว้แล้วในแอป Flink ใหม่ของโปรเจกต์อื่น จึงต้องหาวิธีแทนที่ส่วนการส่งอีเวนต์แบบมีเงื่อนไขและการทำ logic หลังจาก join อีเวนต์แล้ว

เปรียบเทียบวิธีแทนที่

  • หาก implement เป็น Flink App ตัวเดียว จำนวนสิ่งที่ต้องจัดการจะน้อย แต่มีโอกาสสูงที่จะกลับไปเป็นแอปขนาดใหญ่อีกครั้ง และความล้มเหลวในส่วนหนึ่งอาจกระทบฟังก์ชันอื่นได้
  • หากแยกเป็น Flink App หลายตัว จะสามารถจัดการแต่ละแอปได้อย่างอิสระ แต่ยิ่งจำนวนแอปเพิ่มขึ้น ภาระด้าน cluster, resource และ deployment ก็เพิ่มตามไปด้วย
  • Flink SQL ช่วยให้กำหนด logic ด้วย query พัฒนาได้รวดเร็ว และจัดการเพียง cluster เดียว แต่การแสดง logic ที่ซับซ้อนด้วย SQL เพียงอย่างเดียวทำได้ยาก และต้องมีประสบการณ์ด้านการปฏิบัติการ cluster
  • ทีมมีความเข้าใจ implementation ภายในของ Flink มากขึ้นแล้ว และประเมินว่า Flink SQL มีข้อได้เปรียบด้าน ผลิตภาพ และ ประสิทธิภาพการปฏิบัติการ

เหตุผลที่เลือก Flink SQL

  • Flink SQL ช่วยให้ implement แอปประมวลผล event streaming ด้วย SQL ได้โดยไม่ต้องเขียนโค้ดแอปพลิเคชันโดยตรง
  • ในแง่ High Availability (HA) Flink รองรับการประมวลผลแบบมี state และสามารถบันทึก/กู้คืนสถานะของงานเป็นระยะหรือ ณ จุดเวลาที่ต้องการได้ด้วย Checkpoint และ Savepoint
    • JobManager สามารถตั้งค่าเป็นโหมด HA แบบ leader-standby ได้
    • หาก TaskManager บางส่วนล้มเหลว สามารถกระจายงานของ TaskManager ที่ล้มเหลวไปยัง TaskManager อื่นได้ตามกลยุทธ์ retry ของ Job
  • แค่ syntax ของ SQL ก็สามารถจัดการฟังก์ชันหลักของการประมวลผลสตรีมมิงได้
    • แปลงรูปแบบข้อมูลด้วย SELECT และกรอง record ด้วย WHERE
    • รวมหลาย stream ด้วย JOIN และรวม stream เข้าด้วยกันด้วย UNION
    • รองรับการประมวลผลแบบ window เช่น tumbling, hopping(sliding), session window
    • สามารถกำหนดขอบเขตการยอมรับข้อมูลที่มาช้าด้วย event time processing และ watermark
  • ขยายข้อกำหนดที่อยู่นอกเหนือฟังก์ชันพื้นฐานด้วย UDF และ custom connector
    • ส่วนใหญ่ของระบบเลกาซีเดิมเป็นแพตเทิร์นที่ใช้คำสั่ง Redis SET หรือ INCR และเนื่องจากไม่มี Redis Connector อย่างเป็นทางการของ Flink จึงเขียน Redis Connector เองและนำมาใช้
    • ในตอนนั้นไม่มี built-in function สำหรับหาส่วนร่วมของชนิด ARRAY จึง implement ด้วย UDF แล้วนำไปใช้ใน query

เปรียบเทียบกับ ksqlDB และ Spark Structured Streaming

  • ksqlDB รวมอยู่ในแพลตฟอร์ม Confluent ที่บริษัทใช้สำหรับ Kafka และมี use case ระดับบริษัทอยู่แล้ว
  • อย่างไรก็ตาม ทีมประเมินว่ามีความไม่มีประสิทธิภาพในการทำงานแบบ HA ของการประมวลผลสตรีมมิงที่มี state
    • เมื่อ failover ของ stateful operation ต้อง replay changelog ซึ่งเป็นบันทึกการเปลี่ยนแปลง state ทั้งหมด จึงอาจใช้เวลา failover นาน
    • วิธีที่มี replica ของ processing stream และอัปเดต changelog เข้า internal state อย่างต่อเนื่อง จะทำ operation เดียวกันบน replica ด้วย ทำให้อาจใช้ resource เป็นสองเท่า
    • ดูรายละเอียดที่เกี่ยวข้องได้ที่ Configuring ksqlDB for High Availability | Confluent Developer
  • Spark Structured Streaming เป็น engine ประมวลผลสตรีมมิงที่อิง Spark SQL engine
    • มี use case ภายในบริษัท และสามารถเขียน UDF กับ Custom Sink ได้
    • มี ecosystem ที่ใหญ่และพัฒนามาดีกว่า Flink
  • Spark ทำงานเป็นหน่วย micro-batch จึงอาจเกิด latency ในระดับ record และในสถานการณ์ที่ real-time processing สำคัญ อาจเสียเปรียบ Flink
  • ภายในทีมแทบไม่มีประสบการณ์ Spark และยังต้องเขียน Custom Sink ด้วย จึงเลือก Spark ได้ไม่ง่ายนัก

การสร้างสภาพแวดล้อม cluster

  • ในเครื่อง local สามารถดาวน์โหลด binary จาก หน้าเว็บทางการของ Flink แล้วรัน cluster ด้วย {FLINK_HOME}/bin/start-cluster.sh
  • เมื่อรัน {FLINK_HOME}/bin/sql-client.sh จะเปิด Flink SQL CLI และสามารถส่ง test query เช่น SELECT 1; ได้
  • หลังส่ง query แล้ว สามารถตรวจสอบใน Flink web UI ได้ว่า query ที่ส่งถูกแปลงเป็น Job และรันอยู่
  • ช่วงปลายปี 2022 Flink SQL Gateway ถูก release ทำให้สามารถส่ง query ผ่าน HTTP ได้

สถาปัตยกรรมปฏิบัติการบน Kubernetes

  • เนื่องจากบริการส่วนใหญ่ภายในบริษัททำงานบน Kubernetes จึงจัด Flink SQL Cluster บน Kubernetes ด้วย
  • Flink App เดิมทั้งหมดถูก deploy และ operate ในรูปแบบ Application mode
    • เป็นวิธีที่รัน cluster แยกสำหรับแต่ละ application
    • บน Kubernetes แต่ละแอปจะรัน JobManager Pod และ TaskManager Pod แยกกัน
    • มีข้อดีด้านความเป็นอิสระและการแยกส่วนระหว่างแอป รวมถึงการจัดการ config และ dependency ตามแต่ละงาน
  • Flink SQL ส่ง Job เข้า cluster ที่รันอยู่แล้ว ดังนั้นต้องรัน JobManager และ TaskManager ใน Session mode
  • Cluster ถูกจัดตามคู่มือ Stand Alone Cluster on Kubernetes
  • สภาพแวดล้อม HA อ้างอิง config จาก High-Availability with Standalone Kubernetes และใช้ s3 เป็น high-availability.storageDir
  • วิธี Native Kubernetes รัน cluster ด้วย shell script ที่ให้มา จึงประเมินว่าไม่เหมาะกับสภาพแวดล้อม infrastructure ภายในบริษัทที่กำหนด deployment config เองแล้ว deploy

การตั้งค่าเชื่อมต่อ HA และ S3

  • เพื่อเชื่อมต่อ HA และ S3 ใช้ config ต่อไปนี้ใน 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}
  • ในสภาพแวดล้อม HA จะรัน JobManager pod สองตัว และ address ของทั้งสองต้องแตกต่างกัน logic การเลือก leader ฯลฯ จึงจะทำงานได้ปกติ
  • ตั้งค่า argument สำหรับรัน container ของ JobManager ดังนี้
args: ["start-foreground", "-D", "jobmanager.rpc.address=$(POD_IP)"]
  • ด้วย config นี้ ข้อมูลของ JobManager pod ที่ถูกเลือกเป็น leader ปัจจุบันและ Job ID ที่กำลังรันอยู่จะถูกบันทึกไว้ใน Kubernetes ConfigMap และนำไปใช้กับ HA

Deploy query แบบ GitOps

  • ปัจจุบัน Flink ยังไม่มี web UI หรือเครื่องมือเฉพาะสำหรับ Flink SQL ที่ให้มาโดยตรง
  • เคยพิจารณา use case การเชื่อมต่อ Hue เป็น PoC แต่ในตอนนั้นมีปัญหา compatibility กับเวอร์ชันของ Flink SQL Gateway ต้องพัฒนาเพิ่มเติม และการตั้งค่าสภาพแวดล้อมสำหรับพัฒนาก็ใช้เวลามาก
  • ภายในบริษัทใช้ GitOps pattern กันมาก จึง implement GitHub Actions สำหรับ deploy query หรือหยุด Job
  • สร้างโฟลเดอร์ตาม Job ใน Repository และรวบรวม query ที่จะรันไว้เป็นไฟล์ SQL
  • GitHub Actions รับชื่อโฟลเดอร์และระบุไฟล์ SQL ที่จะดึง query ออกมา
  • การ implement ใช้วิธีเรียก Flink SQL Gateway REST API และเขียนด้วย Python ซึ่งเรียบง่ายและทดสอบได้ง่าย

ตัวอย่างการปฏิบัติการและการรับมือเหตุขัดข้อง

  • ยังไม่เคยพบ JobManager ล้มเหลว แต่ตาม config HA หาก JobManager ล้มเหลว JobManager ตัวอื่นจะถูกเลือกเป็น leader และทำงานต่อได้
  • TaskManager ล้มเหลวเป็นครั้งคราว และส่วนใหญ่เป็นกรณีที่ Pod ถูก restart ตาม policy Kubernetes QoS
  • ยืนยันแล้วว่าแม้ TaskManager บางส่วนล้มเหลว งานก็ถูกกระจายใหม่ไปยัง TaskManager อื่นและดำเนินต่อได้
  • Query failure ส่วนใหญ่เกิดจาก ข้อมูลผิดปกติที่ไหลเข้ามา หรือ ทรัพยากร compute ไม่เพียงพอ
    • เมื่ออ่านข้อมูล JSON สามารถละเว้นข้อมูล error จาก JSON format ที่ไม่ถูกต้องได้ด้วย option json.ignore-parse-errors
    • error ที่เกิดเมื่อดึงข้อมูลจาก path บางจุดด้วย JSON_VALUE แล้วไม่มีค่าหรือชนิดไม่ตรง สามารถกำหนดค่า default ด้วย DEFAULT {VALUE} ON ERROR
    • หาก CPU ของ TaskManager เกิน 100% หรือ memory ไม่พอ จะเพิ่ม resource ของ TaskManager หรือเพิ่ม parallelism ของ query แล้ว redeploy
  • มีบางครั้งที่ Job บางตัวล้มเหลวเมื่อ restart cluster เนื่องจากเปลี่ยน config ของ cluster หรือเพิ่ม UDF
    • สาเหตุมักเป็น timeout หรือ retry config ของ Job ที่ไม่เหมาะสม
    • ปรับ timeout และ retry config เพื่อไม่ให้ Job หยุด retry เร็วเกินไป และให้ retry ต่อจนกว่า cluster จะเสถียรหลัง restart

ข้อจำกัดของการเปลี่ยน query และการกู้คืน state

  • เมื่อแก้เงื่อนไข query แล้ว deploy ใหม่ กรณีที่สามารถกู้คืน state ด้วย savepoint ได้จำกัดอยู่ที่การแก้ไขที่เรียบง่ายมาก เช่น เปลี่ยนค่าของ expression เงื่อนไข
  • หากเงื่อนไข window เปลี่ยน state ก็จะเปลี่ยนตาม ทำให้รักษา compatibility ได้ยาก และอาจกู้คืนด้วย savepoint ได้ยาก
  • หากจำเป็นต้องรักษา state แต่ requirement เปลี่ยนบ่อย การเขียนแอปโดยตรงอาจเหมาะกว่า

จุดที่ควร monitor

  • Flink มี metric ที่ให้มาโดยพื้นฐานจำนวนมาก หากมี monitoring infrastructure ภายในบริษัทและ Metric Reporter ที่เหมาะสม ก็สามารถสร้างสภาพแวดล้อม monitoring ได้ง่าย
  • numRunningJobs แสดงจำนวน Job ที่กำลังรันอยู่ใน cluster ปัจจุบัน หากค่าลดลงกะทันหันและค้างอยู่เช่นนั้น สามารถพิจารณาได้ว่ามี Job ที่ล้มเหลว
  • ใช้ taskmanager.cpu.load และ taskmanager.memory.used เพื่อดูการใช้ resource ของ cluster
  • ใช้ busyTimeMsPerSecond เพื่อดูว่า TaskManager ยุ่งแค่ไหนในแต่ละ Job
  • หากใช้ Kafka เป็น source สามารถตรวจสอบสถานะ data lag ได้อย่างรวดเร็วด้วย records-lag-max

ตัวอย่าง: การ aggregate window ของอีเวนต์ login จาก Kafka

  • ตัวอย่างใน Appendix รับอีเวนต์จาก Kafka แล้วส่ง จำนวน login event ในช่วง 1 นาทีที่ผ่านมา ทุก 10 วินาที ไปยัง Kafka
  • ข้อมูล input เป็น JSON format และมีฟิลด์ event_time, event_type, data.user_id
  • Query ตั้งค่า pipeline.name, parallelism.default, table.exec.state.ttl
  • ตาราง input login_event ใช้ Kafka connector และ JSON format พร้อมตั้ง json.ignore-parse-errors เป็น true
  • row_time สร้างจาก event_time และตั้ง watermark ให้ประมวลผลอีเวนต์ที่มาช้าได้สูงสุด 5 วินาที จาก event time สูงสุดที่สังเกตพบจนถึงปัจจุบัน
  • ตาราง output windowed_login_count ส่งผลลัพธ์ไปยัง Kafka topic และมีฟิลด์ proc_time AS PROCTIME()
  • ใช้ HOP(row_time, INTERVAL '10' SECOND, INTERVAL '1' MINUTE) เพื่อสร้าง hopping window ความยาว 1 นาที ทุก 10 วินาที และ aggregate จำนวน record ใน window ด้วย COUNT(*)

ผลลัพธ์การปฏิบัติการและจุดปรับปรุงที่เหลือ

  • ทีมสามารถใช้ประสบการณ์ Flink เดิมเพื่อเพิ่มฟังก์ชันหลายอย่างได้ง่ายและเร็วกว่าก่อน
  • ประเมินว่าได้ผลลัพธ์ที่น่าพอใจในแง่ผลิตภาพและประสิทธิภาพการปฏิบัติการ
  • หลังนำมาใช้ ทำงานได้อย่างเสถียรเป็นเวลาประมาณ 1 ปี โดยแทบไม่ต้องมีงานปฏิบัติการเพิ่มเติม
  • ปัจจุบันกำลังค่อย ๆ ขยายการใช้งาน
  • การ redeploy query และการเปลี่ยน config ของ cluster ยังมีความไม่สะดวกอยู่ และวางแผนปรับปรุงสภาพแวดล้อมการ deploy query ผ่านการ implement GitOps Controller pattern

บทความที่เกี่ยวข้อง

1 ความคิดเห็น

 
flgkselql98 2025-02-26

ระบบแบบกระจายอย่าง flink จำเป็นต้องคงไว้ซึ่ง HA ด้วยการมี rack 2~3 ชุด และดูเหมือนว่าพอผนวกกับ kubernetes แล้วก็เหมือนจะรับประกัน HA ได้ แต่สุดท้ายก็คงต้องคิดเรื่องทรัพยากรของ kube slave node อยู่ดี เลยสงสัยว่าเขาจัด node ที่รันแต่ flink แยกไว้หรือเปล่า (ถ้า flink มีโหลดสูง ก็น่าจะมีปัญหา slave node ล่มได้)
ในมุมแบบนั้น การใช้ kubernetes มีข้อดีอะไรบ้างไหม?

อีกอย่าง ถ้าใช้ window function ใน flink ข้อมูลระหว่างนั้นจะถูกเก็บไว้ในหน่วยความจำ ทำให้คำสั่ง SQL join ทำงานได้ แต่ถ้ามองในแง่ trade-off ก็เลยทำให้คิดว่า flink เป็นตัวเลือกที่ดีจริงหรือเปล่า ถ้าเวลาผ่านไปแล้ว SQL + job ขนาดใหญ่ขึ้นเรื่อยๆ แล้ว job ตายขึ้นมา ผลกระทบก็คงมหาศาล..

ผมเองก็สงสัยเหมือนกันว่า ถ้าอยู่ในสถานการณ์ที่ต้อง join กันตั้งแต่ data source ชั้นบนสุด จะมีวิธีไหนที่ไม่ใช้ flink แล้วลดระดับมาจัดการที่ application level ได้บ้าง.