Retour d’expérience sur l’adoption de Flink SQL

• L’Azar Matching Dev Team de Hyperconnect a choisi d’implémenter le traitement de streaming en SQL plutôt qu’en code applicatif, afin de découper une application legacy Flink monolithique qui utilisait 96 CPU
• Découper en plusieurs applications Flink améliore l’isolation, mais augmente la charge d’exploitation ; l’équipe a donc estimé que Flink SQL, qui permettait de tirer parti de sa compréhension interne de Flink, était plus adapté en matière de productivité et d’efficacité opérationnelle
• Flink SQL prend en charge Checkpoint/Savepoint, la HA du JobManager, la redistribution des TaskManagers, les fenêtres, les jointures, l’event time, les watermarks, les UDF et les connecteurs personnalisés ; il correspondait donc mieux à la situation de l’équipe que ksqlDB et Spark Structured Streaming
• L’environnement de production a été construit sous forme de cluster Flink en Session mode sur Kubernetes, et le déploiement des requêtes ainsi que l’arrêt des jobs ont été gérés en mode GitOps avec l’API Flink SQL Gateway et GitHub Actions
• Après environ un an d’exploitation stable, l’usage est en cours d’extension, mais le redéploiement des requêtes et les changements de configuration du cluster restent peu pratiques ; une amélioration fondée sur le pattern GitOps Controller est prévue

---

Contexte du remplacement d’une lourde application legacy de streaming

• L’Azar Matching Dev Team gérait plusieurs applications basées sur Flink, dont une lourde application legacy utilisant 96 CPU
• Cette application avait une architecture monolithique qui regroupait au même endroit plusieurs fonctionnalités : jointure de plusieurs événements de matching, publication conditionnelle d’événements, stockage de flags Redis, etc.
• Après un changement des nœuds d’exécution lié à des travaux d’infrastructure à l’échelle de l’entreprise, l’application ne fonctionnait plus correctement, et un simple tuning ne permettait pas de résoudre rapidement le problème
• Comme la fonctionnalité critique de jointure d’événements avait déjà été implémentée dans une nouvelle application Flink d’un projet distinct, il fallait remplacer la partie qui publiait conditionnellement des événements et exécutait la logique après la jointure des événements

Comparaison des approches de remplacement

• Une implémentation sous forme d’une seule application Flink réduit le nombre d’éléments à gérer, mais risque fortement de recréer une application énorme, et la défaillance d’une partie peut affecter d’autres fonctionnalités
• Un découpage en plusieurs applications Flink permet de gérer chaque application indépendamment, mais plus le nombre d’applications augmente, plus la charge liée aux clusters, aux ressources et aux déploiements augmente également
• Flink SQL permet de définir la logique par des requêtes, de développer rapidement et de ne gérer qu’un seul cluster, mais il est difficile d’exprimer uniquement en SQL une logique complexe, et une expérience d’exploitation de cluster est nécessaire
• L’équipe avait acquis une bonne compréhension de l’implémentation interne de Flink et a estimé que Flink SQL offrait des avantages en matière de productivité et d’efficacité opérationnelle

Pourquoi Flink SQL a été choisi

• Flink SQL permet d’implémenter des applications de traitement de flux d’événements en SQL sans écrire directement de code applicatif
• Du point de vue de la haute disponibilité (HA), Flink prend en charge le traitement stateful, et Checkpoint et Savepoint permettent de sauvegarder et restaurer l’état des traitements périodiquement ou à un instant voulu
  • Le JobManager peut être configuré en mode HA de type leader-standby
  • Si une partie des TaskManagers échoue, les tâches du TaskManager défaillant peuvent être redistribuées vers d’autres TaskManagers selon la stratégie de retry du Job
• La syntaxe SQL suffit à couvrir les principales fonctionnalités de traitement de streaming
  • Transformer la forme des données avec SELECT et filtrer les enregistrements avec WHERE
  • Combiner plusieurs streams avec JOIN et fusionner des streams avec UNION
  • Prendre en charge le fenêtrage, comme les tumbling windows, hopping windows (sliding) et session windows
  • Définir la tolérance aux données en retard avec le traitement en event time et les watermarks
• Les UDF et connecteurs personnalisés permettent d’étendre les besoins au-delà des fonctionnalités de base
  • La plupart des traitements du legacy existant suivaient un pattern utilisant les commandes Redis SET ou INCR, et comme il n’existait pas de Redis Connector officiel pour Flink, l’équipe a écrit et utilisé son propre Redis Connector
  • À l’époque, il n’existait pas de fonction intégrée pour calculer l’intersection d’un type ARRAY ; elle a donc été implémentée sous forme d’UDF et utilisée dans les requêtes

Comparaison avec ksqlDB et Spark Structured Streaming

• ksqlDB était inclus dans la plateforme Confluent utilisée en interne pour Kafka, et il existait aussi des cas d’usage à l’échelle de l’entreprise
• Cependant, l’équipe a estimé qu’il présentait des inefficacités dans le comportement HA du traitement de streaming stateful
  • Lors du failover d’une opération stateful, il faut rejouer l’intégralité du changelog, c’est-à-dire l’historique des changements de state, ce qui peut allonger le temps de failover
  • L’approche consistant à maintenir une réplique du stream de traitement et à mettre à jour en continu le changelog dans le state interne peut consommer deux fois plus de ressources, car la réplique effectue les mêmes opérations
  • Les détails sont disponibles dans Configuring ksqlDB for High Availability | Confluent Developer
• Spark Structured Streaming est un moteur de traitement de streaming basé sur le moteur Spark SQL
  • Il existait des cas d’usage en interne, et il permet d’écrire des UDF et des Custom Sinks
  • Il dispose d’un écosystème plus vaste et mieux établi que Flink
• Spark fonctionne par micro-batchs, ce qui peut introduire une latence au niveau des enregistrements ; dans les situations où le traitement temps réel est important, cela peut le désavantager par rapport à Flink
• L’équipe avait très peu d’expérience Spark, et comme il fallait aussi écrire un Custom Sink, il était difficile de choisir Spark sans hésitation

Mise en place de l’environnement de cluster

• En local, il est possible de télécharger les binaires depuis le site officiel de Flink et de lancer le cluster avec {FLINK_HOME}/bin/start-cluster.sh
• En exécutant {FLINK_HOME}/bin/sql-client.sh, la CLI Flink SQL s’ouvre et permet de soumettre une requête de test comme SELECT 1;
• Après soumission de la requête, l’interface web de Flink permet de vérifier que la requête soumise a été transformée en Job et exécutée
• Fin 2022, la sortie de Flink SQL Gateway a rendu possible la soumission de requêtes via HTTP

Architecture d’exploitation basée sur Kubernetes

• Comme la plupart des services internes fonctionnent sur Kubernetes, le cluster Flink SQL a lui aussi été configuré sur Kubernetes
• Les applications Flink existantes étaient toutes déployées et exploitées en Application mode
  • Cela consiste à lancer un cluster séparé pour chaque application
  • Sur Kubernetes, chaque application fonctionne avec ses propres pods JobManager et TaskManager
  • C’était avantageux pour l’indépendance et l’isolation entre applications, ainsi que pour la gestion des configurations et des dépendances propres à chaque job
• Comme Flink SQL soumet des Jobs à un cluster déjà lancé, il fallait démarrer le JobManager et les TaskManagers en Session mode
• Le cluster a été configuré sur la base du guide Stand Alone Cluster on Kubernetes
• Pour l’environnement HA, l’équipe s’est référée à la configuration High-Availability with Standalone Kubernetes, avec s3 comme high-availability.storageDir
• L’approche Native Kubernetes lance le cluster via un shell script fourni ; elle a donc été jugée inadaptée à l’environnement d’infrastructure interne, où les configurations de deployment sont définies et déployées directement

Configuration HA et intégration S3

• Pour la HA et l’intégration S3, les paramètres suivants ont été utilisés dans 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}

• Dans un environnement HA, deux pods JobManager sont lancés ; leurs adresses doivent être différentes pour que la logique d’élection du leader fonctionne correctement
• Les arguments d’exécution du conteneur JobManager sont configurés comme suit

args: ["start-foreground", "-D", "jobmanager.rpc.address=$(POD_IP)"]

• Grâce à cette configuration, les informations du pod JobManager élu leader et les ID des Jobs actuellement en cours d’exécution sont stockés dans un ConfigMap Kubernetes et utilisés pour la HA

Déploiement des requêtes en mode GitOps

• Flink ne fournit pas encore d’interface web ou d’outil dédié à Flink SQL
• L’intégration avec Hue a été évaluée en PoC, mais des développements supplémentaires étaient nécessaires à l’époque en raison de problèmes de compatibilité de version avec Flink SQL Gateway, et la mise en place de l’environnement de développement prenait aussi beaucoup de temps
• Comme le pattern GitOps est largement utilisé en interne, l’équipe a implémenté des GitHub Actions pour déployer les requêtes ou arrêter les Jobs
• Dans le repository, un dossier est créé par Job, et les requêtes à exécuter sont regroupées dans des fichiers SQL
• GitHub Actions reçoit le nom du dossier et identifie le fichier SQL dont extraire la requête
• L’implémentation appelle la Flink SQL Gateway REST API et a été écrite en Python, un choix simple et facile à tester

Cas d’exploitation et gestion des incidents

• L’équipe n’a jamais connu de défaillance du JobManager, mais avec la configuration HA, si un JobManager échoue, un autre JobManager peut être élu leader et poursuivre le travail
• Les TaskManagers échouaient parfois ; il s’agissait le plus souvent de pods redémarrés à cause de la politique QoS de Kubernetes
• Il a été confirmé que même lorsqu’une partie des TaskManagers échoue, les tâches sont redistribuées vers d’autres TaskManagers et continuent de s’exécuter
• Les échecs de requêtes sont principalement dus à l’ingestion de données anormales ou à un manque de ressources de calcul
  • Lors de la lecture de données JSON, l’option json.ignore-parse-errors permet d’ignorer les données en erreur si le format JSON est invalide
  • Lors de l’extraction de données à un path donné avec JSON_VALUE, les erreurs dues à l’absence de valeur ou à un type différent peuvent être traitées en définissant une valeur par défaut avec DEFAULT {VALUE} ON ERROR
  • Si le CPU des TaskManagers dépasse 100 % ou si la mémoire est insuffisante, les ressources des TaskManagers sont augmentées ou le parallelism de la requête est accru, puis la requête est redéployée
• Certains Jobs échouaient parfois lors du redémarrage du cluster après un changement de configuration du cluster ou l’ajout d’UDF
  • La cause était souvent une configuration inadaptée du timeout ou des retries du Job
  • Le timeout et les paramètres de retry ont été modifiés afin que le Job n’arrête pas ses retries trop rapidement et continue jusqu’à ce que le cluster se stabilise après le redémarrage

Contraintes liées aux changements de requêtes et à la restauration du state

• Lorsqu’une requête est modifiée puis redéployée, la restauration du state via savepoint n’est possible que pour des modifications très simples, comme un changement de valeur dans une condition
• Si les conditions de window changent, le state change aussi, ce qui rend difficile le maintien de la compatibilité et peut empêcher une restauration avec savepoint
• Si le state doit être conservé alors que les exigences changent fréquemment, écrire directement une application peut être une meilleure approche

Points de monitoring

• Flink fournit de nombreux metrics par défaut ; avec une infrastructure de monitoring interne et un Metric Reporter approprié, il est facile de mettre en place un environnement de monitoring
• numRunningJobs indique le nombre de Jobs actuellement en cours d’exécution dans le cluster ; si la valeur baisse brusquement et reste basse, on peut considérer qu’un Job a échoué
• taskmanager.cpu.load et taskmanager.memory.used permettent de comprendre l’utilisation des ressources du cluster
• busyTimeMsPerSecond permet de vérifier, par Job, à quel point les TaskManagers sont occupés
• Lorsque Kafka est utilisé comme source, records-lag-max permet de vérifier rapidement l’état de retard des données

Exemple : agrégation fenêtrée d’événements de connexion Kafka

• L’exemple en annexe reçoit des événements depuis Kafka et publie dans Kafka, toutes les 10 secondes, le nombre d’événements de connexion sur la dernière minute
• Les données d’entrée sont au format JSON et contiennent les champs event_time, event_type et data.user_id
• La requête configure pipeline.name, parallelism.default et table.exec.state.ttl
• La table d’entrée login_event utilise le connector Kafka et le format JSON, avec json.ignore-parse-errors défini à true
• row_time est généré à partir de event_time, et le watermark est configuré pour traiter les événements arrivant avec un retard maximal de 5 secondes par rapport à l’event time maximal observé jusque-là
• La table de sortie windowed_login_count publie le résultat dans un topic Kafka et inclut le champ proc_time AS PROCTIME()
• HOP(row_time, INTERVAL '10' SECOND, INTERVAL '1' MINUTE) crée une hopping window d’une minute avec un pas de 10 secondes, et COUNT(*) agrège le nombre d’enregistrements dans la window

Résultats d’exploitation et améliorations restantes

• L’équipe a pu exploiter son expérience Flink existante pour ajouter plusieurs fonctionnalités plus facilement et plus rapidement qu’auparavant
• Elle estime avoir obtenu un résultat satisfaisant en matière de productivité et d’efficacité opérationnelle
• Depuis son adoption, le système fonctionne de manière stable depuis environ 1 an, sans opérations d’exploitation particulières
• L’exploitation est actuellement progressivement étendue
• Le redéploiement des requêtes et les changements de configuration du cluster restent peu pratiques, et l’équipe prévoit d’améliorer l’environnement de déploiement des requêtes en implémentant le pattern GitOps Controller

[flgkselql98]

Les systèmes distribués comme Flink doivent maintenir la HA en conservant 2 à 3 racks, et j’ai l’impression que cela a été garanti en l’intégrant à Kubernetes. Mais au final, il faut aussi réfléchir aux ressources des nœuds workers kube, donc je me demande s’ils ont constitué des nœuds dédiés à Flink uniquement (en cas de forte charge Flink, il semble qu’il puisse y avoir des problèmes de panne des nœuds workers).
Dans cette optique, y a-t-il vraiment un avantage à utiliser Kubernetes ?

De plus, quand on utilise des fonctions de fenêtre dans Flink, les données intermédiaires sont conservées en mémoire pour faire fonctionner les jointures SQL ; du point de vue des compromis, je me demande si Flink est vraiment un bon choix. Le problème énorme qui survient si un SQL + job qui grossit avec le temps finit par tomber...

Moi aussi, dans les cas où une jointure est nécessaire tout en haut de la data source, je me demande comment on pourrait traiter cela au niveau applicatif sans utiliser Flink, en le faisant descendre à l’application.