Grâce à ses millions de capteurs disséminés dans toutes ses installations, Veolia collecte chaque jour des centaines de gigaoctets de données brutes. Ces données exploitées sous différents aspects permettent à l’entreprise de prédire les pannes et la consommation, détecter les taux anormaux ou encore de procéder aux télérelevés. L’objectif du Datalake, plateforme qui unifie et agrège des données en provenance de différentes sources, est d’avoir un point d’entrée unique pour l’ensemble des données métier, tout en étant capable de filtrer, agréger des milliards de lignes.  L’énorme quantité de données à traiter a amené à élaborer une architecture élastique basée sur BigQuery et Dataflow, facilement scalable en conservant une maîtrise des coûts.

Première version avec Airflow

Historiquement Airflow était utilisé pour collecter et stocker les données. C’est un ordonnanceur de tâches, il permet de programmer des traitements (scripts python) à la manière crontab. Il fournit une interface de monitoring qui permet de suivre le statuts des jobs.

Pour son fonctionnement il a besoin d’une VM pour s’exécuter, d’une base Redis et d’une base MySQL. Le principal problème d’Airflow était le manque de scalabilité. L’ajout de nouveaux workers est un tâche manuelle qui nécessite l’ajout d’une VM dédiée au pool Celery (Message Queue).

La nécessité d’un certain nombre de VM et le manque de scalabilité font qu’il était de plus en plus difficile d’utiliser Airflow pour ces opérations. Chaque instance supplémentaire est une VM qu’il faut manager, monitorer et la charge de travail croît continuellement.

Dataflow le successeur

Dans l’optique de construire une architecture serverless, D2SI a participé à l’optimisation et à l’industrialisation du datalake. Le passage de Airflow vers Dataflow a apporté une grande flexibilité. Dataflow est un service de Google Cloud Platform qui permet d’exécuter un pipeline Apache Beam (framework Python/Java de manipulation de données) sans avoir à gérer un cluster. Lorsqu’un pipeline s’exécute il crée à la volée un groupe d’instances allant de 1 à 1000 VM (bloqué à 1000 dans les quotas par défaut). En fonction de la quantité de données traitée, le groupe d’instances scale automatiquement.

Le datalake collecte de manière régulière des données brutes principalement en provenance de serveurs FTP/SFTP, mais également d’API REST de buckets Amazon s3 ou encore par envoi d’email. L’utilisation de Dataflow permet, à partir de code Python en utilisant le framework Apache Beam de générer automatiquement un processus d’ingestion de données.

Pour gérer les différents cas et processus d’ingestion de données, le pipeline de traitement est généré à partir d’une configuration en JSON qui décrit la source de données. Cette configuration est stockée dans Datastore.

Elle inclut des paramètres tels que:

• La fréquence d’ingestion
• Le schéma de la table de destination
• Le support par lequel les données sont récupérées (FTP/SFTP/API/BUCKET)
• Liste d’opérations à appliquer (filtres, anonymisation …)

Par ailleurs le système de Messaging PubSub est utilisé afin d’assurer l’ordonnancement des fichiers et leur traitement optimal en quasi temps réel par les jobs Dataflow. Nous avons développé une série de fonctionnalités complémentaires au framework, qui permettent d’appliquer des transformations aux données et d’isoler les données non conformes.

La configuration permet d’orchestrer le pipeline en aiguillant les données de bloc en bloc jusqu’à la dernière étape, l’insertion dans BigQuery. La configuration est stockée dans Datastore, gratuit en dessous d’un certain volume, il n’est que peu sollicité dans ce workflow.

Les jobs sont déclenchés par un CronJob qui en appelant une API (App Engine Flexible) déclenche la génération du pipeline Dataflow.

Bilan

En suivant la politique et les bonnes pratiques du groupe, le datalake a été migré et re-achitecturé pour devenir 100% serverless. Dès le début du projet un ensemble de bonnes pratiques ont été mises en place. Chaque ajout de fonctionnalité passe obligatoirement par une “pull request” sur github, ce qui permet au code d’être audité avant son intégration dans le projet, chaque “pull request” doit être validée et approuvée par au moins un développeur pour être mergée.

Le déploiement est également automatisé par Travis, chaque “merge” de “pull request” sur les branches de production et de pré-production. Avant de procéder au déploiement Travis vérifie la qualité du code en exécutant 800+ tests unitaires, de même la couverture du code est calculée et son résultat stocké dans BigQuery, ce qui permet à l’équipe de visualiser l’évolution de la qualité du code.

Le datalake ingère plus de 30 sources de données différentes, ce qui représente un total d’environ 30 milliards de lignes. Le Serverless a ici permis de répondre aux enjeux de scalabilité du projet, tout en offrant plus de flexibilité par rapport aux approches traditionnelles. L’architecture Serverless a également permis de réduire considérablement les coûts de build et de run, et assure à Veolia de répondre à ses contraintes de disponibilité et de sécurité, comme nous l’explique Sébastien Morand, Team Lead Solution Architect Veolia :

“Mon objectif était d’industrialiser le Datalake. Plutôt que de consacrer de l’énergie à créer from scratch un connecteur pour chaque nouvelle source, l’idée était d’avoir un système léger, via des fichiers de configuration, permettant d’intégrer des sources hétérogènes rapidement. Cela demandait de créer une couche d’abstraction, tout en gardant un système autoscalé, automanagé et No ops aligné avec la stratégie groupe. C’est dans cette optique que nous avons créé la v2 du Datalake. La v1 n’était pas complètement Serverless et comportait des briques IaaS, or nous voulions tendre autant que possible vers le No ops, notamment pour réduire les coûts grâce au pay-as-you-use, d’où l’idée de construire cette nouvelle version industrialisée et totalement serverless.

Ce choix s’est avéré payant, la plateforme nous permet d’avoir des coûts non linéaires par rapport à l’usage et aux nombres de sources. Ce dernier a été multiplié par 5, mais les coûts sont restés stables : le coût du run n’est pas proportionnel au nombre de sources. Par ailleurs, le choix du serverless nous a permis de monter un projet opérationnel, capable d’agréger des milliards de lignes, sans effort de gestion opérationnelle, avec une équipe restreinte de 3 développeurs.

Ce projet s’inscrit complètement dans la stratégie Veolia Datacenterless, qui vise à choisir des solutions managées pour se concentrer sur les couches métiers plutôt que sur les couches basses. Par ailleurs, les briques serverless sont « secure by design« , et le paiement à l’usage permet de réelles économies d’échelle. Non seulement le serverless réduit le coût de l’innovation, mais il permet d’accélérer le time-to-market des réponses aux besoins métiers. »