Comment feriez vous pour transférer 1 600 000 comptes utilisateurs d’un système vers un autre ? Alors que vous n’avez pas de système d’ingestion en masse mais seulement des APIs REST qui prennent un seul compte à la fois. Comment rajouter une nouvelle donnée sur plusieurs millions de documents pour une évolution fonctionnelle ? Chez Groupe La Centrale il nous arrive de devoir faire des scripts de migration, d’import ou de rattrapage de données. Que ce soit parce que l’on souhaite transférer des données d’un système à un autre, parce que l’on doit corriger une donnée quelque part ou parce que l’on a changé un format et que l’on souhaite rendre d’anciens documents compatibles afin de simplifier le code.
Tous ces besoins ont en commun
- ils n’ont pas forcément vocation à être utilisé très souvent (peut-être seulement une fois)
- ils peuvent être amenés à devoir manipuler une quantité relativement importante de données
- ils peuvent causer une charge importante sur les systèmes de production au moment de leur(s) exécution(s)
- on veut avoir une idée du temps d’exécution et être capable de l’optimiser (sans écraser la prod)
Parmi les solutions qui existent, j’aimerais vous en présenter l’une d’entre elles que j’aime beaucoup. Il s’agit d’Akka Streams.
DISCLAIMER : ce n’est bien évidemment pas la seule solution pour ce type de besoins, et ce n’est sans doute pas toujours la solution la plus adaptée (je pense notamment aux cas où la volumétrie est trop faible ou au contraire trop importante). Mais pour bien des cas que nous avons rencontrés, Akka Streams a parfaitement fait l’affaire.
Parmi les points forts d’Akka Streams je noterais
- Basé sur Scala un langage de programmation que j’aime beaucoup (au secour les scripts bash qu’on commence à écrire et où on finit par mettre de la gestion d’erreur pour sauver les meubles)
- Aussi disponible en Java (même si je n’ai jamais essayé)
- Basé sur Akka un framework Scala bien connu
- Un écosystème d’extensions (Alpakka, contrib) très complet et très très pratique
- Le système de throttle et de backpressure (lié à Reactive Streams) qui permet très simplement de maîtriser le niveau de charge qu’on va générer
- Un système à base de « briques lego » que l’on peut combiner très naturellement les unes avec les autres (on finit par avoir une bibliothèque de petites briques que l’on réutilise un peu partout)
- Et par conséquent c’est extrêmement simple à tester
- Le système de retry et de gestion d’erreurs
Cet article n’a pas comme objectif de vous apprendre Akka Streams, néanmoins quelques explications peuvent vous permettre de mieux comprendre de quoi il s’agit.
Akka Streams est construit autour de plusieurs « building blocks » qu’on peut ensuite assembler les uns aux autres.
Les Sources
Ce sont des composants qui vont pouvoir générer un ou plusieurs éléments dans le temps. On notera notamment le type « à gauche » qui correspond aux éléments produits
// Source qui produira 100 entiers de 1 à 100
val source: Source[Int, NotUsed] = Source(1 to 100)
Les Flows
Ce sont des composants qui prennent des éléments en entrée et produisent des éléments en sortie. Attention il n’y a pas forcément une relation 1 <-> 1 entre les éléments en entrée et en sortie.
// Flow qui transforme des entiers en string
val flow1: Flow[Int, String, NotUsed] = Flow[Int]
.map(_.toString)
// Flow qui ne laisse passer que des entiers pairs
val flow2: Flow[Int, Int, NotUsed] = Flow[Int]
.filter(_ != 2)
Les Sink
Ce sont des composants qui consomment des éléments et peuvent produire une valeur finale à la fin
// Sink qui consomme des entiers et produit
// un Future[Seq[Int]]
val sink: Sink[Int, Future[Seq[Int]]] = Sink.seq[Int]
Composition
Là où ça devient très intéressant c’est que l’on peut connecter ces différents éléments les uns aux autres.
Par exemple, si on connecte une Source avec un Flow on obtiendra une nouvelle Source !
val source = Source(1 to 3)
val flow = Flow[Int].map(_.toString)
val newSource: Source[String, NotUsed] = source
.via(flow)
Vous pouvez également connecter 2 Flow pour faire un nouveau Flow, ou bien un Flow et un Sink pour faire un nouveau Sink !
A partir de là, il suffit de regarder la liste des connecteurs proposés dans Alpakka et tout de suite un nombre impressionnant de possibilités s’offrent à nous !
Chez Groupe La Centrale
Par exemple, nous utilisons beaucoup Amazon DynamoDB, qui a un système de scaling où l’on choisit le nombre d’écritures (WCU) et/ou de lectures (RCU) que l’on veut. On sera ensuite facturé sur cette base (il existe aussi un mode On Demand, plus cher, qui ne nécessite pas de choisir nos RCU/WCU).
Dès lors on a envie de s’adapter au mieux à la capacité afin d’en profiter un maximum sans devoir bidouiller notre code dans tous les sens pour faire en sorte d’aller plus ou moins vite. On pourra vouloir lancer à 100 de WCU en local et à 500 en prod par exemple. Akka Streams est parfaitement adapté à ce genre de problématiques avec notamment l’opérateur throttle.
Par ailleurs Akka Streams étant un outil de streaming, toutes les étapes sont effectuées en parallèle. Par exemple : si on lit des éléments depuis une base de données afin de faire des update dessus, on pourra à la fois récupérer un nouveau batch depuis la DB et en même temps faire les écritures du batch précédent.
Déploiement
Se pose ensuite la question du déploiement du « batch » que l’on va vouloir exécuter. Plusieurs options s’offrent à nous, entre autre la bonne vieille EC2 sur laquelle on a installé Java et sur laquelle on peut copier notre jar et l’exécuter. Ça fonctionne très bien. Et en fonction de la nature du batch il est très possible que la majorité du travail consiste à faire des requêtes sur le réseau. Nous avons fait des miracles avec une seule instance T2.SMALL qui réussissait sans problème à DDOS des systèmes distant.
Déploiement @Scale
Si on veut distribuer notre traitement depuis plusieurs machines, plusieurs options s’offrent à nous. Chez Groupe La Centrale nous utilisons beaucoup AWS Lambda et nous avons donc cherché comment intégrer les deux.
L’utilisation de Lambda apporte quelques limitations (temps d’execution max à 15 minutes, impossible de ‘killer’ une lambda en cours d’exécution, …) mais permet d’avoir un scaling en nombre « d’instances » et de déploiement du code très simple et que nous maîtrisons bien.
Pour peu qu’on conçoive notre traitement de façon à pouvoir le découper en plusieurs exécutions (notamment en « stockant » l’état d’avancement) et qu’on le rende « shardable » (avec plusieurs segments en parallèle), on peut facilement mettre tout ça dans une Step Function qui pourra lancer plusieurs instances de notre lambda en parrallèle si on le souhaite.
A partir de là on choisit quel throttle mettre sur chaque process individuellement + combien de process on veut au total et on peut très facilement choisir précisément le niveau de charge qu’on va induire (en passant d’un facteur 1 à 1000 juste avec quelques changements de conf).
Pour aller (encore) plus loin, on peut contourner les limitations de Lambda en utilisant par exemple Izanami, la solution de feature flipping par la MAIF et que nous utilisons (cf l’article de @saziri : ).
On peut ainsi se brancher sur les configs d’Izanami et être notifié de changements qui pourraient être fait et adapter notre stream en conséquence. On peut par exemple ajouter un arrêt d’urgence, ou bien même changer les throttle en live. Malheureusement au cours de mes tests je ne suis pas parvenu à avoir de résultats satisfaisant. La communication entre notre instance Izanami et les lambdas semblait assez aléatoire (j’ai certainement dû faire une erreur dans le système de souscription aux changements de config).
Note : je n’inclus pas de chiffres sur les performances d’Akka Streams justement car dans aucun de nos cas nous n’avons été limité par ses performances. Soit le traitement se faisait dans un temps qui nous convenait, soit il fallait d’abord augmenter les capacités des systèmes en face. Beaucoup de nos systèmes sont serverless et c’est tout à fait faisable, mais on se retrouve généralement avec quelque part dans la chaîne soit un DynamoDB sur lequel on doit mettre 2 000 WCU, un système legacy qui pose problème, un partenaire externe qui ne comprend pas pourquoi on lui envoie 10 fois le trafic habituel, … . Une grande force d’Akka Streams résidant justement dans le fait de maîtriser aisément le débit que l’on produit.
Pour vous donner malgré tout une idée, la migration dont il était question en début d’article et qui consistait à migrer 1 600 000 comptes utilisateurs d’un ancien système vers un nouveau a pris moins de 45 minutes (depuis une seule instance T2.SMALL).
Conclusion
Avec Akka Streams nous avons pu créer un grand nombre de scripts de migration mais aussi des processus qui tournent tous les jours en prod, en voici quelques uns
- dépiler des messages d’une queue SQS afin de les dédupliquer et les renvoyer dans une nouvelle queue
- dépiler des queues SQS pour faire des écritures dans DynamoDB tout en s’assurant de ne pas consommer 100% des capacités afin que d’autres tâches puissent continuer à s’exécuter (et en associant de l’autoscaling ça fait des merveilles)
- relire tous les zips contenant des fichiers json d’un bucket S3 afin de modifier une propriété dans certains documents et réuploader le fichier derrière
- scanner toutes les annonces de notre table DynamoDB afin de calculer leur côte et calculer leur badge bonne affaire lorsque nous avons mis en place la feature
- et bien d’autres !
J’espère que cet article vous aura intéressé(e) et qu’il vous incitera à regarder ce super outil !
Vous pourrez retrouver les sources du projet qui a service d’exemple sur github : https://github.com/cbm-gplassard/article-akkastream
Et vous, comment vous assurez vous de faire tourner des tâches de la façon la plus efficace possible sans écrouler la production au passage ?
Sources et références
Groupe La Centrale Tech - AWS et Agilité 