Einführung von Flink SQL

 (hyperconnect.github.io)
3 Punkte von GN⁺ 2025-02-25 | 1 Kommentare | Auf WhatsApp teilen
  • Das Azar Matching Dev Team von Hyperconnect entschied sich dafür, Streaming-Verarbeitung statt mit Anwendungscode per SQL zu implementieren, um eine monolithische Legacy-App auf Flink-Basis aufzuteilen, die 96 CPUs nutzte
  • Eine Aufteilung in mehrere Flink-Apps verbessert zwar die Isolierung, erhöht aber auch den Betriebsaufwand. Daher kam das Team zu dem Schluss, dass Flink SQL, mit dem sich das interne Verständnis von Flink im Team nutzen lässt, hinsichtlich Produktivität und Betriebseffizienz besser geeignet ist
  • Flink SQL unterstützt Checkpoint/Savepoint, JobManager-HA, Neuverteilung von TaskManagern, Fenster·Joins·Event Time·Watermarks sowie UDFs und benutzerdefinierte Connectoren und passte damit besser zur Teamsituation als ksqlDB und Spark Structured Streaming
  • Die Betriebsumgebung wurde als Flink-Cluster im Session mode auf Kubernetes aufgebaut, und mit der Flink SQL Gateway API sowie GitHub Actions werden Query-Deployments und das Stoppen von Jobs per GitOps umgesetzt
  • Das System läuft seit rund einem Jahr stabil und wird schrittweise erweitert, doch Query-Re-Deployments und Änderungen an den Cluster-Einstellungen sind weiterhin umständlich, weshalb Verbesserungen auf Basis des GitOps-Controller-Patterns geplant sind

Hintergrund für den Ersatz der schweren Legacy-Streaming-App

  • Das Azar Matching Dev Team verwaltete mehrere Flink-basierte Apps, darunter eine schwere Legacy-App, die 96 CPUs nutzte
  • Diese App hatte eine monolithische Struktur, in der Funktionen wie das Joinen mehrerer Match-Events, bedingtes Veröffentlichen von Events und das Speichern von Redis-Flags an einer Stelle gebündelt waren
  • Nachdem im Rahmen unternehmensweiter Infrastrukturarbeiten die Ausführungs-Nodes geändert worden waren, funktionierte die App nicht mehr korrekt, und eine schnelle Lösung war allein durch einfaches Tuning schwer zu erreichen
  • Da die wichtige Event-Join-Funktion bereits in einer neuen Flink-App eines separaten Projekts implementiert war, wurde ein Ansatz benötigt, der den Teil mit bedingter Event-Veröffentlichung und Logikausführung nach dem Event-Join ersetzt

Vergleich der Ersatzansätze

  • Eine Implementierung als eine einzige Flink-App reduziert zwar die Zahl der zu verwaltenden Objekte, birgt aber ein hohes Risiko, wieder zu einer riesigen App zu werden, und ein Fehler in einem Teil kann andere Funktionen beeinträchtigen
  • Bei einer Aufteilung in mehrere Flink-Apps kann jede App unabhängig verwaltet werden, doch mit wachsender App-Zahl steigen auch Aufwand für Cluster, Ressourcen und Deployment
  • Mit Flink SQL lässt sich die Logik per Query definieren, schnell entwickeln und nur ein Cluster verwalten, allerdings ist komplexe Logik schwer allein mit SQL auszudrücken, und Erfahrung im Cluster-Betrieb ist erforderlich
  • Das Team hatte inzwischen ein hohes Verständnis der internen Flink-Implementierung aufgebaut und bewertete Flink SQL daher als vorteilhaft in Bezug auf Produktivität und Betriebseffizienz

Warum Flink SQL gewählt wurde

  • Mit Flink SQL lassen sich Apps zur Event-Streaming-Verarbeitung per SQL implementieren, ohne Anwendungscode direkt schreiben zu müssen
  • Im Hinblick auf Hochverfügbarkeit (HA) unterstützt Flink zustandsbehaftete Verarbeitung, und mit Checkpoints und Savepoints können Job-Zustände regelmäßig oder zu gewünschten Zeitpunkten gespeichert und wiederhergestellt werden
    • Der JobManager kann im HA-Modus in einer Leader-Standby-Struktur konfiguriert werden
    • Wenn ein Teil der TaskManager ausfällt, können die Aufgaben des ausgefallenen TaskManagers gemäß der Job-Retry-Strategie auf andere TaskManager neu verteilt werden
  • Schon mit der SQL-Syntax lassen sich zentrale Funktionen der Streaming-Verarbeitung abdecken
    • Mit SELECT wird die Datenform transformiert und mit WHERE werden Records gefiltert
    • Mit JOIN lassen sich mehrere Streams verbinden und mit UNION Streams zusammenführen
    • Fensterverarbeitung wie tumbling, hopping(sliding) und session window wird unterstützt
    • Mit Event Time und Watermarks lässt sich der Toleranzbereich für verspätete Daten festlegen
  • Mit UDFs und benutzerdefinierten Connectoren lassen sich Anforderungen außerhalb der Grundfunktionen erweitern
    • Der Großteil des bestehenden Legacy-Systems nutzte Muster mit Redis-SET- oder INCR-Kommandos. Da es keinen offiziellen Redis Connector für Flink gab, wurde ein Redis Connector direkt selbst geschrieben und verwendet
    • Damals gab es keine Built-in-Funktion, um die Schnittmenge des ARRAY-Typs zu berechnen, daher wurde dies als UDF implementiert und in Queries genutzt

Vergleich mit ksqlDB und Spark Structured Streaming

  • ksqlDB war in der im Unternehmen für Kafka genutzten Confluent-Plattform enthalten, und es gab dafür unternehmensweite Use Cases
  • Allerdings wurde das HA-Verhalten bei zustandsbehafteter Streaming-Verarbeitung als ineffizient bewertet
    • Beim Failover zustandsbehafteter Operationen muss das gesamte Changelog, also die Aufzeichnung der State-Änderungen, erneut abgespielt werden, wodurch das Failover lange dauern kann
    • Der Ansatz, Replikate des verarbeitenden Streams vorzuhalten und das Changelog fortlaufend in den internen State zu übernehmen, kann den Ressourcenverbrauch verdoppeln, weil auch die Replikate dieselben Operationen ausführen
    • Details dazu finden sich unter Configuring ksqlDB for High Availability | Confluent Developer
  • Spark Structured Streaming ist eine Streaming-Engine auf Basis der Spark-SQL-Engine
    • Es gibt unternehmensinterne Use Cases, und UDFs sowie Custom Sinks können geschrieben werden
    • Es besitzt ein größeres und besser ausgebautes Ökosystem als Flink
  • Spark arbeitet in Micro-Batches, wodurch Latenz auf Record-Ebene entstehen kann, was in Situationen mit hoher Echtzeitanforderung gegenüber Flink nachteilig sein kann
  • Da im Team kaum Spark-Erfahrung vorhanden war und zudem ein Custom Sink geschrieben werden musste, fiel die Wahl auf Spark nicht leicht

Aufbau der Cluster-Umgebung

  • Lokal kann das Binary von der offiziellen Flink-Webseite heruntergeladen und der Cluster mit {FLINK_HOME}/bin/start-cluster.sh gestartet werden
  • Führt man {FLINK_HOME}/bin/sql-client.sh aus, öffnet sich die Flink-SQL-CLI, in der sich Test-Queries wie SELECT 1; absenden lassen
  • Nach dem Absenden einer Query lässt sich in der Flink-Web-UI prüfen, dass die eingereichte Query in einen Job umgewandelt und ausgeführt wurde
  • Mit dem Release von Flink SQL Gateway Ende 2022 wurde HTTP-basiertes Absenden von Queries möglich

Kubernetes-basierte Betriebsarchitektur

  • Da die meisten unternehmensinternen Services auf Kubernetes laufen, wurde auch der Flink-SQL-Cluster auf Kubernetes aufgebaut
  • Die bestehenden Flink-Apps wurden alle im Application mode bereitgestellt und betrieben
    • Dabei wird pro Anwendung ein separater Cluster gestartet
    • Auf Kubernetes laufen pro App jeweils ein JobManager-Pod und TaskManager-Pods
    • Das war vorteilhaft hinsichtlich Unabhängigkeit und Isolierung zwischen Apps sowie bei Job-spezifischen Einstellungen und der Verwaltung von Abhängigkeiten
  • Da Flink SQL Jobs an einen bereits laufenden Cluster sendet, mussten JobManager und TaskManager im Session mode betrieben werden
  • Der Cluster wurde auf Basis des Leitfadens Stand Alone Cluster on Kubernetes aufgebaut
  • Für die HA-Umgebung wurde die Konfiguration aus High-Availability with Standalone Kubernetes herangezogen, und als high-availability.storageDir wurde S3 verwendet
  • Der Ansatz Native Kubernetes wurde als ungeeignet für die interne Infrastruktur bewertet, da er den Cluster über bereitgestellte Shell-Skripte startet, während intern Deployments mit direkt definierten Deployment-Einstellungen verwendet werden

HA- und S3-Integrationskonfiguration

  • Für HA und die S3-Integration wurden in config.yaml die folgenden Einstellungen verwendet
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}
  • In einer HA-Umgebung werden zwei JobManager-Pods gestartet, und ihre Adressen müssen sich unterscheiden, damit die Leader-Election-Logik korrekt funktioniert
  • Die Startargumente des JobManager-Containers werden wie folgt gesetzt
args: ["start-foreground", "-D", "jobmanager.rpc.address=$(POD_IP)"]
  • Mit dieser Konfiguration werden in der Kubernetes-ConfigMap Informationen über den aktuell zum Leader gewählten JobManager-Pod und die aktuell laufenden Job-IDs gespeichert und für HA genutzt

Query-Deployment im GitOps-Stil

  • Es gibt bislang keine von Flink direkt bereitgestellte Web-UI oder dedizierten Tools speziell für Flink SQL
  • Ein PoC mit Hue wurde geprüft, erforderte damals aber wegen Versionskompatibilitätsproblemen mit Flink SQL Gateway zusätzliche Entwicklung, und auch der Aufbau der Entwicklungsumgebung kostete viel Zeit
  • Da im Unternehmen häufig das GitOps-Pattern verwendet wird, wurden GitHub Actions implementiert, um Queries zu deployen oder Jobs zu stoppen
  • Im Repository wurden für jeden Job Ordner angelegt und die auszuführenden Queries als SQL-Dateien gesammelt
  • GitHub Actions nehmen den Ordnernamen entgegen und bestimmen darüber die SQL-Datei, aus der Queries extrahiert werden
  • Die Implementierung ruft die Flink SQL Gateway REST API auf und wurde in Python geschrieben, weil das einfach und gut testbar ist

Betriebserfahrungen und Störungsbehebung

  • Ein Ausfall des JobManagers wurde bisher nicht erlebt, doch laut HA-Konfiguration kann bei einem JobManager-Ausfall ein anderer JobManager zum Leader gewählt werden und die Arbeit fortsetzen
  • TaskManager fielen gelegentlich aus, meist weil Pods aufgrund von Kubernetes-QoS-Richtlinien neu gestartet wurden
  • Es wurde bestätigt, dass auch beim Ausfall einzelner TaskManager die Arbeit auf andere TaskManager umverteilt wird und weiterläuft
  • Query-Fehler traten meist wegen fehlerhafter eingehender Daten oder unzureichender Rechenressourcen auf
    • Beim Lesen von JSON-Daten können fehlerhafte JSON-Formate mit der Option json.ignore-parse-errors ignoriert werden
    • Fehler beim Extrahieren von Daten über einen bestimmten Pfad mit JSON_VALUE, etwa wenn ein Wert fehlt oder der Typ nicht passt, lassen sich mit DEFAULT {VALUE} ON ERROR durch einen Standardwert abfangen
    • Wenn die TaskManager-CPU über 100 % liegt oder nicht genug Speicher vorhanden ist, werden die TaskManager-Ressourcen erhöht oder die Query-Parallelität gesteigert und anschließend neu deployt
  • Beim Neustart des Clusters wegen geänderter Cluster-Konfigurationen oder zusätzlicher UDFs schlugen manche Jobs fehl
    • Die Ursache waren oft ungeeignete Job-Timeout- oder Retry-Einstellungen
    • Daher wurden Timeout- und Retry-Einstellungen so angepasst, dass Jobs die Wiederholungsversuche nicht zu früh beenden, sondern bis zur Stabilisierung nach dem Cluster-Neustart weiter retryen

Einschränkungen bei Query-Änderungen und State-Wiederherstellung

  • Wenn Query-Bedingungen geändert und erneut deployt werden, ist eine Wiederherstellung des State per Savepoint nur bei sehr einfachen Änderungen möglich, etwa beim Anpassen von Werten in Bedingungen
  • Ändern sich Window-Bedingungen, ändert sich auch der State, wodurch Kompatibilität schwer aufrechtzuerhalten ist und eine Wiederherstellung per Savepoint schwierig werden kann
  • Wenn der State erhalten bleiben muss, sich die Anforderungen aber häufig ändern, kann ein direkt geschriebener App-Ansatz besser geeignet sein

Wichtige Monitoring-Punkte

  • Flink bietet viele integrierte Metriken, sodass sich mit einer passenden internen Monitoring-Infrastruktur und einem geeigneten Metric Reporter leicht eine Monitoring-Umgebung aufbauen lässt
  • numRunningJobs zeigt die Zahl der aktuell im Cluster laufenden Jobs; fällt der Wert plötzlich ab und bleibt dort, kann von einem fehlgeschlagenen Job ausgegangen werden
  • Mit taskmanager.cpu.load und taskmanager.memory.used lässt sich die Ressourcenauslastung des Clusters erfassen
  • Über busyTimeMsPerSecond lässt sich pro Job prüfen, wie stark ein TaskManager ausgelastet ist
  • Wenn Kafka als Source verwendet wird, kann mit records-lag-max der Zustand von Datenverzögerungen schnell überprüft werden

Beispiel: Fensteraggregation von Kafka-Login-Events

  • Das Beispiel im Appendix empfängt Events aus Kafka und veröffentlicht alle 10 Sekunden die Anzahl der Login-Events der vergangenen 1 Minute wieder nach Kafka
  • Die Eingabedaten liegen im JSON-Format vor und enthalten die Felder event_time, event_type und data.user_id
  • Die Query setzt pipeline.name, parallelism.default und table.exec.state.ttl
  • Die Eingabetabelle login_event verwendet den Kafka-Connector und das JSON-Format, wobei json.ignore-parse-errors auf true gesetzt ist
  • row_time wird aus event_time erzeugt, und der Watermark ist so konfiguriert, dass Events verarbeitet werden, die bis zu 5 Sekunden später als die bislang beobachtete Event Time eintreffen
  • Die Ausgabetabelle windowed_login_count veröffentlicht die Ergebnisse in ein Kafka-Topic und enthält das Feld proc_time AS PROCTIME()
  • Mit HOP(row_time, INTERVAL '10' SECOND, INTERVAL '1' MINUTE) wird ein 1-Minuten-Hopping-Window im 10-Sekunden-Abstand erzeugt, und mit COUNT(*) wird die Zahl der Records innerhalb des Windows aggregiert

Betriebsergebnisse und verbleibende Verbesserungen

  • Das Team konnte seine bestehende Flink-Erfahrung nutzen, um mehrere Funktionen einfacher und schneller als zuvor hinzuzufügen
  • Das Ergebnis wurde hinsichtlich Produktivität und Betriebseffizienz als zufriedenstellend bewertet
  • Nach der Einführung lief das System etwa 1 Jahr lang stabil, ohne nennenswerte Betriebsarbeiten zu erfordern
  • Der Betrieb wird derzeit schrittweise ausgeweitet
  • Beim erneuten Deployment von Queries und bei Änderungen an Cluster-Konfigurationen bestehen weiterhin Unannehmlichkeiten, und es ist geplant, die Query-Deployment-Umgebung durch die Implementierung eines GitOps-Controller-Patterns zu verbessern

Verwandte Beiträge

1 Kommentare

 
flgkselql98 2025-02-26

Verteilte Systeme wie Flink müssen für HA typischerweise 2–3 Racks vorhalten; durch die Anbindung an Kubernetes scheint HA hier sichergestellt worden zu sein. Allerdings muss man sich am Ende doch auch Gedanken über die Ressourcen der Kubernetes-Worker-Nodes machen. Da frage ich mich, ob dafür Nodes konfiguriert wurden, auf denen nur Flink läuft (bei hoher Flink-Last dürfte es wohl Probleme geben, wenn ein Worker-Node ausfällt).
Aus dieser Perspektive: Welche Vorteile hat der Einsatz von Kubernetes?

Wenn man in Flink außerdem Window-Funktionen verwendet, bleiben die Daten in dieser Zeit im Speicher, sodass SQL-Joins funktionieren. Unter Trade-off-Gesichtspunkten frage ich mich daher, ob Flink wirklich eine gute Wahl ist. Wenn ein immer größer werdendes SQL + Job mit der Zeit abstürzt, ist das schon eine enorme Sache ...

Ich überlege ebenfalls, wie man in Situationen, in denen bereits an der obersten Data Source Joins notwendig sind, das auf Application-Ebene herunterziehen und verarbeiten könnte, statt Flink zu verwenden.