Designing Cloud Data Platforms

1 - Introducing the data platform

Les analytics permettent essentiellement d'obtenir des métriques pour faire des choix business.
- Avant l'avènement des ordinateurs, les entreprises utilisaient des moyens manuels, et leur intuition.
- Dans les années 80 on a vu émerger le concept de data warehouse, qui est une base centralisée de données venant de diverses sources.
Les data warehouses posent de plus en plus de problèmes de nos jours.
- Les tendances suivantes y contribuent :
  - Les données sont issues de sources de diverses nature, y compris certaines d'entre-elles non structurées, et leur volume est de plus en plus important.
  - Le découpage des applications en microservices fait que collecter des données revient forcément à devoir agréger de multiples sources.
  - Les data scientists ont aussi besoin d'accéder à une version brute de la donnée, et cet usage ne peut pas passer par un data warehouse.
- Elles ont du mal avec les 3V (Variety, Volume, Velocity).
  - Variety : les data warehouses ne supportent que les structured data dont le schéma est stable, c'est-à-dire en pratique qui sont issues de DB relationnelles.
    - Or avec l'avènement des SaaS, des réseaux sociaux, et de l'IoT, on se retrouve avec :
      - Des semistructured data du type JSON, Avro etc, dont le schéma varie souvent.
      - Des unstructured data comme le binaire, le son, la vidéo.
  - Volume : le fait que dans un data warehouse, la puissance de calcul et le stockage doivent se trouver sur la même machine physique, implique qu'on ne peut pas scaler les deux séparément, et donc les coûts explosent.
    - Même les petites organisations peuvent être amenées à traiter plusieurs TB de données.
  - Velocity : les data warehouses ne sont pas adaptées aux analytics en mode real time, elles sont plus orientées batch processing.
  - Le machine learning en particulier pose tous les problèmes en même temps : il nécessite une grande quantité de données variées, et accapare la puissance de calcul du data warehouse.
Les data lakes répondent en partie à ces problèmes.
- L'idée principale des data lakes c'est qu'on stocke de la donnée telle quelle (ou quasi), et qu'on essayera de la traiter et de lui coller un schéma dès qu'on en aura besoin.
- Les data lakes se sont généralisés à partir de 2006 avec l'arrivée de Hadoop, qui est un filesystem distribué sur plusieurs machines pas chères.
  - Hadoop répond en partie aux 3V :
    - A la Variety par l'écriture schema-less.
    - Au Volume par le fait que ce soit distribué sur des machines pas chères.
    - A la Velocity par la facilité de streaming à partir du filesystem distribué.
  - Mais il a aussi des problèmes :
    - C'est un système complexe qu'il faut installer sur un datacenter et gérer par des Ops expérimentés.
    - D'un point de vue business, c'est plus difficile de travailler avec les outils qui traitent les données non structurées qu'avec du SQL comme dans un data warehouse.
    - Bien qu'il soit distribué sur de petites machines pas chères, le computing et le stockage ne sont pas séparés, ce qui limite quand même la réduction de coût quand on a besoin de beaucoup de l'un sans l'autre.
- Le cloud public vient répondre aux problèmes de Hadoop.
  - Les data warehouses et les data lakes ont été proposés par les cloud providers, avec de nombreux avantages :
    - La possibilité de scaler la puissance de calcul et le stockage séparément.
    - Payer uniquement à l'usage des machines qu'on emprunte.
    - Ne plus avoir à gérer la complexité de l'infrastructure.
    - Des outils et frameworks avancés développés par les cloud providers autour de leurs produits.
  - Exemple : AWS EMR permet de lancer un cluster sur lequel on va pouvoir exécuter des jobs Hadoop et Spark,
    - On a juste à indiquer le nombre de nœuds qu'on veut, et les packages qu'on veut installer dessus.
    - Et on a la possibilité de faire des allers-retours vers S3 pour scaler différemment le calcul et le stockage.
La cloud data platform moderne utilise à la fois le data warehouse et le data lake, hébergés dans un cloud public, chacun d'entre eux remplissant un usage particulier.
- Pour être polyvalente et pas chère, la data platform doit avoir des 4 composants principaux faiblement couplés, interagissant entre-eux avec une API bien définie.
  - Ingestion layer : on va chercher les données chez les différents types de sources (DB relationnelle, DB NoSQL, API externes etc.).
    - On va en général utiliser un ensemble d'outils open source ou commerciaux pour chaque type de données à aller chercher.
    - Il ne faut surtout pas altérer la donnée à cette étape, pour que la donnée brute soit disponible pour les data scientists qui en auraient l'usage.
  - Storage layer : on utilise le stockage cloud comme stockage de notre data lake, dans lequel on met ce qu'on a ingéré.
    - Le stockage cloud a l'avantage de ne pas avoir besoin de planifier la capacité de stockage : il grossit automatiquement au besoin.
  - Processing layer : on transforme la donnée pour la rendre utilisable par la plupart des clients de la plateforme.
    - C'est la partie calcul de notre data lake, il va lire depuis le cloud storage puis écrire à nouveau dedans.
    - Dans le cas du streaming, on ne passe pas par le storage layer qui prend trop de temps, mais on envoie la donnée directement au processing layer, qui va ensuite la rendre disponible au layer d'après.
    - Le processing est généralement fait avec des outils open source, les plus connus étant Spark, Beam et Flink.
  - Serving layer : on rend la donnée disponible sous divers formats, selon les besoins des clients de la plateforme.
    - Les usages peuvent être :
      - Des analystes qui ont besoin d'exécuter des requêtes SQL sur la donnée.
        
        On peut charger la donnée dans un data warehouse chez le cloud provider.
      - Des applications qui ont besoin d'un accès rapide à la donnée.
        
        On peut la charger dans une key / value DB, ou une document DB.
      - Des équipes de data scientists / engineers ont besoin de transformer la donnée eux-mêmes.
        
        On peut leur donner accès au storage du data lake, et les laisser utiliser Spark, Beam ou Flink.
- La cloud data platform répond aux 3V :
  - L'ingestion layer couplé au stockage sans schéma permet une grande Variety des données.
  - La séparation calcul / stockage et le fait de ne payer que ce qu'on utilise permet d'optimiser les coûts, et d'avoir un gros Volume.
  - La possibilité d'envoyer directement au processing layer permet de la Velocity.
  - On peut aussi prendre en compte deux autres V :
    - La Veracity qui indique le niveau de data governance, c'est-à-dire la qualité de la donnée. On l'obtient itérativement, au cours d'étapes au sein du data lake.
    - Et la Value qu'on peut tirer de la donnée, qui peut être plus élevée si on prend plus de données en amont de notre processus de nettoyage.
Il faut comprendre les cas d'usages principaux d'un data lake, pour éviter de le transformer en data swamp. Parmi les plus courants il y a :
- La vue 360° des clients, où il s'agit de récupérer toutes les données d'interaction avec eux, pour proposer ensuite des services plus personnalisés, vendre plus etc.
- Les données venant d'IoT, qui ont la particularité d'être incertaines et d'avoir un gros volume, ce qui rend l'utilisation du data warehouse peu intéressante.
- Le machine learning qui a besoin d'une très grande quantité de données, et qui tire avantage de puissance de calcul séparée des autres use-cases grâce au data lake.

2 - Why a data platform and not just a data warehouse

Ce chapitre donne des arguments pour le choix d'une cloud data platform, plutôt qu'une simple data warehouse.
On implémente les deux solutions pour une situation d'exemple qu'on va utiliser dans ce chapitre :
- Nous sommes l'équipe data, et le département marketing a besoin que nous récupérions deux sources de données et qu'on les corrèle régulièrement.
  - L'une des sources est une table de campagnes de marketing, issue d'une DB MySQL interne.
  - Et l'autre est constituée de fichiers CSV de clics utilisateurs, issus de logs applicatifs (et donc semistructured).
- On part sur Microsoft Azure pour les deux solutions.
- Concernant l'implémentation data warehouse only :
  - 1 - On va utiliser deux Azure Data Factory pour récupérer la donnée dans le serveur de DB et les fichiers CSV dans le serveur SFTP. C'est notre ingest layer.
  - 2 - Ensuite on redirige ça vers l'Azure Synapse, qui est la data warehouse de chez Azure. Elle va faire office de store layer, process layer et serve layer.
- Concernant l'implémentation cloud data platform :
  - 1 - On a notre ingest layer avec Azure Data Factory, qui redirige les données vers le store layer.
  - 2 - Le store layer est implémenté avec Azure Blob Storage. Il s'agit d'un stockage de type data lake.
  - 3 - On a un process layer qui utilise Azure Databricks, et qui fait tourner Spark.
  - 4 - Le serve layer enfin utilise Azure Synapse qui est le data warehouse.
Concernant l'ingestion.
- Pour la version data warehouse only :
  - La pipeline contient :
    - Des linked services : ici la data source MySQL en entrée, et la data sink Azure Synapse en sortie.
    - Des data sets : il s'agit de la description du schéma de données d'entrée et de sortie, et leur mapping.
  - Si le schéma de la DB source change, il faudra mettre à jour le schéma défini dans la pipeline et le mapping.
    - Mais surtout il faudra gérer soi-même la migration du data sink.
- Pour la version cloud data platform :
  - Cette fois le data sink est un Azure Blob Storage.
    - Il n'y a plus besoin de spécifier les schémas et le mapping entre input et output puisque l'output accueille la donnée telle quelle.
  - Si le schéma de la DB source change, il n'y a rien à faire côté ingestion : on écrira de toute façon la donnée dans un nouveau fichier.
    - On déplace le problème de mapping plus loin.
Concernant le processing.
- Dans la version data warehouse only :
  - On va charger les deux données :
    - La DB MySQL sans charger sa structure parce qu'elle est déjà relationnelle.
    - La donnée CSV semistructurée dans des rows de type texte qu'on parsera en JSON avec une fonction SQL built-in.
  - La requête SQL qu'on va écrire aura les désavantages suivants :
    - Elle sera peu lisible, à cause du code de parsing nécessaire.
      - On pourrait la rendre plus lisible en pré-parsant la donnée, mais ça veut dire plus de temps et des coûts plus élevés.
      - Une autre solution de lisibilité pourrait être d'ajouter des UDF (User Defined Functions), qu'il faudrait maintenir et déployer sur chaque instance d'Azure Synapse.
    - Elle sera difficile à tester.
    - Elle risque de ne pas profiter de la performance offerte par la structure en colonne du data warehouse, parce que les données texte qu'on parse en JSON ne sont pas organisables physiquement en colonnes.
- Dans la version cloud data platform :
  - On a la possibilité d'utiliser un distributed data processing engine comme Apache Spark.
    - On pourra écrire des requêtes SQL pour des expérimentations rapides.
    - Et on pourra aussi écrire du code lisible, maintenable et testable dans un langage comme Python ou Scala, quand il s'agit de projet de plus long terme.
Concernant l'accès à la donnée.
- Il peut y avoir plusieurs types de consommateurs :
  - Des utilisateurs plutôt orientés business comme des équipes marketing.
    - Ils vont préférer utiliser des outils de reporting type Power BI, et donc auront besoin de la donnée sous forme relationnelle, par exemple dans Azure Synapse.
  - Des utilisateurs orientés data analyse / data science.
    - Ils pourront bénéficier de SQL qu'ils utilisent souvent directement, au travers de Spark SQL.
    - Ils pourront avoir accès à des données non filtrées pour leur projets data science, grâce Spark directement.
  - Au final la cloud data platform, qui contient à la fois la donnée sous forme brute dans le data lake, et la donnée dans le data warehouse, est adaptée à chaque usage.
A propos des coûts financiers.
- Il est difficile de comparer les coûts des services cloud.
  - En général on constate que le stockage est plutôt pas cher, et que l'essentiel des coûts se trouve dans les calculs.
- L'elastic scaling consiste à pouvoir calibrer le service pour l'usage exact qu'on en a, et de ne pas avoir à payer plus.
  - C'est un des éléments qui permet de vraiment optimiser les coûts.
- Pour la version data warehouse only, l'essentiel des coûts va aller dans Azure Synapse.
  - Le scaling de ce service peut prendre des dizaines de minutes, donc c'est quelque chose qu'on ne peut faire que de temps en temps.
- Pour la version cloud data platform, l'essentiel des coûts est porté par le processing layer, par exemple Spark.
  - Spark est particulièrement élastique, au point où il est commun de démarrer une instance juste le temps d'une requête.

3 - Getting bigger and leveraging the Big 3: Amazon, Microsoft, and Google

Il existe un trade off entre choisir des services vendor-specific de type PaaS, et choisir des services open source.
- D'un côté on se couple au vendor mais on minimise les coûts d'Ops, et de l'autre on permet une meilleure portabilité mais on augmente les coûts d'Ops.
- Les auteurs trouvent que la solution vendor-specific est celle qui a en général le moins de désavantages.
Pour répondre aux problématiques de la data moderne, il faut une architecture en 6 couches.
- 1 - Data ingestion layer.
  - Son but est de :
    - Se connecter aux sources et récupérer la donnée dans le data lake sans trop la modifier.
    - Enregistrer des statistiques et un statut dans le metadata repository.
  - Selon les auteurs, il vaut mieux mettre en place à la fois un mécanisme de type batch et un mécanisme de type streaming.
    - L'industrie est en train de se diriger vers le streaming, mais de nombreuses sources externes fournissent la donnée sous un format de type batch avec des éléments groupés, par exemple CSV, JSON, XML.
    - On pourrait utiliser la partie batch pour ingérer des données par petits batchs, et éviter de faire la version streaming. Mais ça créerait de la dette technique parce qu'on finira par avoir besoin du streaming à un moment ou un autre.
    - La lambda architecture consiste à avoir la donnée qui passe à la fois par le mécanisme de batch et par le mécanisme de streaming.
      - Cette duplication était nécessaire parce que le streaming n'était pas fiable dans les débuts de Hadoop, mais ce n'est plus le cas.
      - La cloud data platform ne consiste pas à faire une telle duplication : selon la source, la donnée va passer par le mécanisme de streaming ou de batch.
  - On entend parfois plusieurs choses différentes quand on parle de real time pour des analytics :
    - 1 - La real time ingestion consiste à avoir la donnée disponible pour de l'analyse dès qu'elle arrive.
    - 2 - Le real time analytics consiste à avoir des fonctionnalités d'analytics qui se mettent à jour à chaque arrivée de donnée.
      - Cette dernière est plus difficile à faire, donc il vaut mieux bien clarifier les besoins.
      - Exemple : détection de fraude en temps réel.
- 2 - Storage layer.
  - Son but est de :
    - Stocker la donnée pour du court terme et du long terme.
    - La rendre disponible pour la consommation streaming et la consommation batch.
  - Le slow storage est là pour le mode batch.
    - La donnée y est persistée pour pas cher, grâce à la possibilité de scaler le stockage sans ajouter de capacité de calcul.
    - Par contre les temps d'accès sont grands.
  - Le fast storage est là pour le mode streaming.
    - Il s'agit d'utiliser un outil qui est fait pour l'accès rapide, comme Apache Kafka.
    - Par contre, on n'a en général pas la possibilité de scaler le stockage sans ajouter de puissance de calcul, et donc les coûts sont plus grands.
    - On va donc purger régulièrement la donnée du fast storage, et de la transférer dans le slow storage.
- 3 - Processing layer.
  - Son but est de :
    - Lire la donnée depuis le stockage et y appliquer de la business logic.
    - Persister la donnée modifiée à nouveau dans le stockage pour un usage par les data scientists.
    - Délivrer la donnée aux autres consumers.
  - Il faut un ou plusieurs outils qui permettent de réaliser des transformations de données, y compris avec du calcul distribué.
    - Un exemple peut être Google Dataflow, qui est une version PaaS d'Apache Beam, qui supporte à la fois le mode streaming et le mode batch.
- 4 - Technical metadata layer.
  - Son but est de :
    - Stocker des informations techniques sur chaque layer.
      - Ça peut être les schémas d'ingestion, le statut de telle ou telle étape, des statistiques sur les données ou les erreurs, etc.
    - Permettre à chaque layer d'ajouter/modifier ou consulter des informations.
  - Par exemple, le processing layer peut vérifier dans la technical metadata layer qu'une certaine donnée est disponible pour aller la chercher, plutôt que de demander à l'ingestion layer.
    - Ce qui permet un certain découplage.
  - D'autres exemples peuvent impliquer des usages de monitoring.
  - La business metadata est une autre notion qui peut avoir son layer, mais qui n'est pas explorée dans ce livre.
    - Il s'agit d'identifier l'usage business qui est fait de chaque donnée qu'on récupère des sources, et d'en faire un catalogue.
  - Il n'y a pas vraiment d'outil unique qui permette de remplir ce rôle pour le moment, donc on devra sans doute en utiliser plusieurs.
    - Par exemple Confluent Schema Registry et Amazon Glue peuvent supporter certains des cas d'usages.
- 5 - Serving layer.
  - Son but est de :
    - Servir les consumers qui ont besoin de données relationnelles via une data warehouse.
    - Servir les consumers qui ont besoin de la donnée brute, en accédant directement au data lake.
      - Les data scientistes vont en général vouloir y accéder via le slow storage.
      - Et l'accès via le fast storage va plutôt intéresser les applications qui s'abonnent en mode streaming.
        
        Par exemple un système de recommandation ecommerce en temps réel.
- 6.1 - Orchestration overlay layer.
  - Son but est de :
    - Coordonner l'exécution de jobs, sous la forme d'un graphe de dépendance.
    - Gérer les échecs et les retries.
  - C'est un peu le complément du technical metadata layer pour permettre le faible couplage entre les layers.
  - L'outil le plus connu d'orchestration est Apache Airflow, adopté par Google Cloud Platform sous le nom de Google Composer.
    - AWS et Azure ont quant à eux choisi d'inclure des fonctionnalités d'orchestration dans leur outil d'ETL.
- 6.2 - ETL overlay layer.
  - Son but est de :
    - Prendre en charge les fonctionnalités de certains layers (ingestion, processing, metadata, orchestration) avec peu ou pas de code.
  - On pourrait faire l'ensemble de notre pipeline avec cet outil ETL, la question à se poser c'est : à quel point il est ouvert à l'extension ?
    - On peut vouloir à l'avenir par exemple utiliser un autre outil de processing, ou s'interfacer avec un outil open source.
    - Dans le cas où il y a une incompatibilité avec un usage qu'on a, on peut toujours l'implémenter à part de l'outil ETL.
      - Le problème c'est qu'au bout d'un moment, les usages à côté deviennent aussi complexes que la solution entière sans l'outil ETL, mais avec une architecture spaghetti.
  - Parmi les outils ETL il y a AWS Glue, Azure Data Factory et Google Cloud Data Fusion.
    - Il existe des solutions commerciales non cloud-natives comme Talend et Informatica, mais ce livre se limite au cloud-native et aux outils open source.
Les couches doivent être bien séparées et découplées.
- Une première raison est de pouvoir utiliser les outils les plus adaptés aux besoins de chaque couche.
  - Le cloud bougeant très vite, on voudra sans doute pouvoir changer seulement l'un d'entre eux quand on a une meilleure alternative pour une couche en particulier.
- Une autre raison est qu'on peut avoir plusieurs équipes en charge de la data platform, et il vaut mieux qu'elles ne se gênent pas.
  - Par exemple, on voudra souvent avoir l'ingestion plutôt centralisée, et le processing plutôt en mode libre service pour chaque équipe qui en a besoin.
Les outils pouvant servir dans une des couches de notre plateforme sont classés en 4 catégories (les auteurs les priorisent dans cet ordre) :
- 1 - Solutions cloud-native PaaS d'AWS, GCP ou Azure.
  - Leur avantage principal c'est le gain de temps : on n'a pas à se préoccuper de la compatibilité. On configure très facilement et c'est en prod.
  - Par contre, c'est la solution qui va être la moins extensible : si par exemple un connecteur n'est pas supporté, on aura du mal à l'ajouter.
  - Elle est aussi peu portable, vu qu'on n'a pas les mêmes services d'un cloud provider à un autre.
- 2 - Solutions serverless.
  - Il s'agit de pouvoir déployer son code custom, mais sans avoir à se préoccuper des serveurs, de leur configuration, du scaling etc.
  - C'est une solution intermédiaire d'un point de vue trade-offs sur la flexibilité, la portabilité et le gain de temps.
- 3 - Solutions open-source.
  - Leur avantage c'est c'est la flexibilité et la portabilité maximales, mais de l'autre côté on a à gérer soi-même des VMs dans le cloud donc plus de travail d'Ops.
- 4 - Solutions SaaS commerciales.
  - Elles peuvent avoir un intérêt si elles ont une fonctionnalité non disponible sous forme PaaS ou open source.
- Dans les faits, on va utiliser un mix de solutions des 4 catégories en fonction des layers et des besoins qu'on a.
  - On a de plus en plus d'entreprises qui utilisent des solutions de plusieurs cloud providers. Par exemple le gros des services sur AWS, et le use-case machine learning sur GCP.
Outils sur AWS.
- Batch ingestion.
  - AWS Glue supporte l'ingestion à partir de AWS S3, ou à partir d'une connexion JDBC.
  - AWS Database Migration Service sert à la base à transférer ses DBs vers AWS, mais on peut l'utiliser comme ingestion layer.
  - AWS DMS permet d'implémenter un mécanisme de change data capture à partir d'une DB.
  - Si aucune des solutions PaaS ne supporte notre data source, on peut utiliser la solution serverless AWS Lambda où il faudra écrire et maintenir du code.
- Streaming ingestion.
  - AWS Kinesis est un message broker pour lequel il faudra écrire du code pour publier dedans. Il a malheureusement très peu de connecteurs entrants.
    - En revanche il a des connecteurs sortants appelés Kinesis Firehose, qui permettent par exemple d'envoyer la donnée de Kinesis dans un S3 sous format Parquet.
  - AWS Managed Streaming for Apache Kafka (MSK) est une version de Kafka entièrement managée.
    - On peut l'utiliser à la place de Kinesis, par exemple si on migre une application avec Kafka vers AWS.
- Storage.
  - AWS S3 permet de stocker de la donnée de manière scalable, avec la possibilité de choisir entre plusieurs formules avec des latences plus ou moins grandes.
- Batch processing.
  - AWS Elastic MapReduce (EMR) est une version managée de Spark.
    - On va en général lire la donnée depuis S3, faire le calcul, puis détruire le cluster EMR.
- Streaming processing.
  - AWS Kinesis Data Analytics permet de se brancher sur Kinesis, et de faire du processing en streaming.
  - Si on utilise AWS MSK, on peut brancher dessus Kafka Streams pour le processing en streaming.
- Data warehouse.
  - AWS Redshift est un data warehouse distributé sur plusieurs noeuds.
    - Redshift Spectrum permet de faire des requêtes depuis Redshift pour obtenir des données qui sont en fait sur S3.
      - Il faudra définir des “tables externes”, et la performance de la query sera moins bonne, mais ça permet d'économiser de la place dans le data warehouse.
- Direct access.
  - AWS Athena permet de faire une requête SQL distribuée en utilisant directement la donnée sur S3.
    - On lance l'instance le temps de la requête, puis on détruit l'instance.
- ETL overlay et metadata repository.
  - AWS Glue est un outil d'ETL complet.
    - Il est construit autour de Spark, et possède des templates pour faciliter de nombreuses transformations.
      - Il a aussi des add-ons Spark non-standards, ce qui nuit à la portabilité par rapport à un simple Spark managé.
    - Il maintient un Data Catalog à partir des données disponibles sur S3.
    - Il maintient un ensemble de statistiques sur l'exécution des jobs.
- Orchestration.
  - AWS Step Functions permet de créer des workflows qui mettent en jeu différents services, y compris ceux qui ne seraient pas gérés par Glue comme AWS Lambda avec du code custom.
- Consumers.
  - Pour les outils comme Tableau qui ont besoin d'une connexion JDBC/ODBC qui supporte SQL, elles peuvent se connecter à Redshift ou Athena.
  - Pour du streaming avec faible latence, on peut envoyer la donnée dans un key/value store comme DynamoDB, ou dans une DB comme AWS RDS ou AWS Aurora.
Outils sur GCP.
- Batch ingestion.
  - Cloud Data Fusion est un ETL overlay qui permet d'ingérer des données depuis une DB relationnelle avec JDBC, des fichiers depuis Google Cloud Storage, et même depuis un FTP ou depuis AWS S3.
    - Il est basé sur un projet open source, et donc supporte des connecteurs custom.
  - BigQuery Data Transfer Service permet d'ingérer de la donnée depuis les services SaaS de Google, et depuis des centaines d'autres services SaaS connus grâce à un partenariat avec Fivetran.
    - Par contre, la donnée va directement dans le data warehouse, ce qui ne permet pas vraiment l'architecture modulaire qu'on vise.
  - Cloud Functions représente l'équivalent d'AWS Lambda, avec le désavantage d'avoir une limite de temps d'exécution des fonctions serverless.
- Stream ingrestion.
  - Cloud Pub/Sub est un broker équivalent à AWS Kinesis.
- Storage.
  - Google Cloud Storage est un équivalent à AWS S3.
- Batch processing.
  - Dataproc est un Spark managé équivalent à AWS EMR.
  - Cloud Dataflow est un Apache Beam managé.
    - Beam a l'avantage d'offrir une même API pour le batch processing et le streaming processing, là où Spark ne supporte que le batch mais est une techno plus mature.
- Streaming processing.
  - Cloud Dataflow représente la manière cloud-native de faire du streaming sur GCP.
  - Dataproc avec Spark Streaming peut représenter une alternative, mais il s'agit en fait de micro-batch et non pas de traiter les messages un par un.
    - Les auteurs conseillent Beam, sauf si on a déjà investi en temps ou connaissances sur Spark.
- Data warehouse.
  - BigQuery est un équivalent à AWS Redshift.
    - Il a l'avantage de scaler le nombre de nœuds tout seul.
    - Par contre il a un modèle de facturation basé sur la donnée lue par chaque requête, ce qui peut rendre les coûts difficiles à prédire.
- Direct access.
  - GCP ne propose pas de services pour accéder au data lake directement avec du SQL.
    - On peut éventuellement créer des tables vers de la donnée externe (donc dans le data lake) à partir de BigQuery.
    - On peut aussi utiliser Spark SQL pour identifier et lire de la donnée sur le data lake.
- ETL overlay et metadata repository.
  - Cloud Data Fusion est un ETL overlay équivalent à AWS Glue. Il fournit une UI qui permet de configurer la pipeline.
    - Il met à disposition un moyen d'analyser quelle partie de la pipeline peut affecter telle ou telle donnée.
    - Il met aussi à disposition des statistiques sur l'exécution des jobs.
- Orchestration.
  - Cloud Composer permet de créer des flows d'orchestration entre jobs.
    - Il est basé sur Apache Airflow.
- Consumers.
  - BigQuery n'a pas de connexion JDBC/ODBC pour y connecter un outil BI par exemple.
    - Il a une API REST, et il est directement compatible avec certains outils BI.
  - Si on veut consommer la donnée avec une faible latence, on peut la mettre dans le key/value store Cloud Bigtable.
Outils sur Azure.
- Batch ingestion.
  - Azure Data Factory est un ETL overlay permettant de faire de l'ingestion depuis diverses sources (DB, SaaS externes, S3, GCS etc.).
    - Il est celui qui a le plus de connecteurs comparé à AWS Glue et Cloud Data Fusion.
  - Azure Functions est l'équivalent d'AWS Lambda.
    - Il ne supporte que Java et Python.
- Streaming ingestion.
  - Azure Event Hubs est équivalent à AWS Kinesis.
    - Il a la particularité d'être compatible avec Apache Kafka.
- Storage.
  - Azure Blob Storage est équivalent à AWS S3.
  - Azure Data Lake Storage est une version améliorée qui supporte mieux le calcul distribué avec de grandes quantités de données.
- Batch processing.
  - Pour le batch processing, Azure a choisi de miser sur un partenariat avec Databricks, qui est un service créé par les créateurs de Spark.
    - La version managée de Databricks est disponible sur AWS et Azure, mais elle est celle par défaut sur Azure, donc mieux supportée par son écosystème.
- Streaming processing.
  - Azure Stream Analytics se branche sur Event Hubs et permet de faire du streaming processing.
- Data warehouse.
  - Azure Synapse est le data warehouse d'Azure.
    - Il est entre AWS Redshift et Google BigQuery dans la mesure où il nécessite de choisir la capacité de calcul, mais il scale l'espace disque tout seul.
- Direct access.
  - Azure Databricks est la manière privilégiée d'accéder à la donnée sur le data lake, soit par l'API native de Spark, soit en SQL avec Spark SQL.
- ETL overlay et metadata repository.
  - Azure Data Factory est équivalent à AWS Glue.
    - Il s'intègre parfaitement avec Databricks pour les transformations complexes.
    - Il fournit des métriques sur la data pipeline.
- Orchestration.
  - La partie orchestration des jobs est prise en charge par Azure Data Factory.
- Consumers.
  - Azure Synapse fournit une connexion JDBC/ODBC pour connecter les outils de BI.
    - Azure Databricks fournit la même chose, mais il faut un cluster Spark toujours allumé, ce qui peut coûter cher.
  - Cosmos DB est une DB orientée document où on peut stocker les résultats de processings pour un accès faible latence.
Alternatives commerciales ou open source.
- Certains logiciels open source sont trop difficiles à mettre en place, par exemple un data warehouse distribué comme Apache Druid.
- Batch ingestion.
  - Il existe pas mal d'outils open source et commerciaux qui permettent d'ingérer des données, leur valeur ajoutée étant en général le grand nombre de sources supportées.
  - Apachi NiFi est une solution open source qui supporte de nombreuses sources, et permet d'en ajouter soi-même en Java.
  - Il existe de nombreux outils SaaS commerciaux qui gèrent l'ingestion.
    - Ces outils vont souvent envoyer la donnée directement dans un data warehouse.
    - Il faut bien réfléchir à la problématique de la sécurité.
- Streaming ingestion.
  - Apache Kafka est le principal outil utilisé en dehors d'une solution managée de streaming.
    - Il a l'avantage de pouvoir se connecter à de nombreuses sources avec Kafka Connect, et il a un moyen d'implémenter des applications de streaming avec Kafka Streams.
    - Les raisons de choisir Kafka plutôt qu'une solution cloud-native peuvent être l'investissement qu'on a déjà dans Kafka (par exemple connaissances), ou le besoin de performance nécessitant le fine-tuning du serveur Kafka.
- Orchestration.
  - Apache Airflow est le principal outil utilisé en dehors d'une solution managée d'orchestration.
    - La raison de l'utiliser en mode non managé peut être de profiter de sa flexibilité, avec ses fichiers en Python.

4 - Getting data into the platform

Le layer d'ingestion peut avoir besoin d'ingérer différents types de données :
- 1 - Les bases de données relationnelles.
  - Leurs données sont organisées en colonnes et typées, mais chaque vendor a des types à lui.
    - Il y a donc un mapping à faire entre le type de chaque colonne et notre modèle.
    - Ce mapping va changer régulièrement en fonction des évolutions fonctionnelles des applications qui possèdent ces DBs.
  - Vu que la donnée est normalisée, elle se trouve dans des centaines de tables.
    - Il faudra donc automatiser le mapping pour éviter de le faire à la main.
  - La donnée change régulièrement dans la DB, pour refléter l'état de l'application, elle est volatile.
    - Il faudra donc aller chercher régulièrement les derniers changements.
- 2 - Les fichiers.
  - Les fichiers sont structurés selon divers types de format texte ou binaire (CSV, JSON XML, Avro, Protobuf etc.) qui ne contiennent pas d'information de type.
    - Il faut donc pouvoir supporter le parsing de tous ces formats.
  - Les fichiers ne garantissent aucun schéma, et on voit beaucoup plus souvent des changements dans celui-ci que pour les DB relationnelles.
    - Il faut donc gérer les changements de schéma fréquents.
  - Les fichiers représentent en général de la donnée figée.
    - La nouvelle donnée est écrite dans un autre fichier, donc on se retrouve à devoir ingérer de nombreux fichiers.
- 3 - La donnée SaaS via API.
  - Les données SaaS sont en général disponibles via une API REST, qui renvoie du JSON.
  - Chaque provider a sa propre API, et son propre format. Il faudra donc implémenter la partie ingestion pour chacun des providers.
    - Il faudra faire la validation du schéma à chaque fois.
    - Il faudra la tenir à jour en fonction des changements d'API.
- 4 - Les streams.
  - Les mêmes données peuvent arriver plusieurs fois, donc il faut que notre pipeline puisse gérer les duplicatas.
  - Les events des streams sont immutables, et peuvent être corrigés en ajoutant un autre message modifié au stream.
    - Donc il faut que notre pipeline gère la réconciliation entre plusieurs versions d'un même message.
  - Les données de streaming ont en général un grand volume, donc il faut une infrastructure qui le supporte.
Concernant le cas des bases de données relationnelles.
- Il y a deux moyens d'ingérer de la donnée depuis une DB relationnelle :
  - 1 - L'utilisation de requêtes SQL.
    - Il s'agit d'avoir un composant qui va :
      - 1 - Exécuter la requête vers la DB concernée.
        
        Ca peut être un simple :
        SELECT * FROM table
      - 2 - Récupérer la donnée sous un format qu'il comprend.
      - 3 - Mapper la donnée dans le bon format pour la stocker sur le storage layer.
      - Il y a donc 2 mappings qui se produisent pendant l'opération.
    - Alors que la donnée opérationnelle s'intéresse à l'état actuel (“Quels sont les articles dans le panier ?”), la donnée analytique s'intéresse à l'évolution de l'état dans le temps (“Quels articles ont été ajoutés ou enlevés et dans quel ordre ?”).
      - Il faut donc un moyen pour capturer l'évolution de la donnée dans le temps.
    - Une 1ère solution pour garder l'évolution dans le temps est de faire une full table ingestion.
      - On va récupérer l'ensemble des données d'une table à intervals réguliers, sauver ces snapshots dans le data lake, et les charger dans le data warehouse.
      - Pour en tirer quelque chose, il faut superposer les rows des snapshots dans la même table du data warehouse.
        
        Pour différencier les rows de chaque snapshot, on peut ajouter une colonne INGEST_DATE.
        
        On peut directement utiliser du SQL pour obtenir les données qu'on veut, mais pour certains usages on aura besoin de faire une transformation dans le processing layer.
      - Parmi les données dérivées qu'on voudra créer, il peut y avoir :
        
        Créer une view qui ne montre que les rows du dernier snapshot.
        
        De la donnée qui identifie les suppressions, en identifiant les rows qui existaient dans un snapshot et n'existaient plus dans le suivant.
        
        Une version “compactée”, qui élimine les rows qui n'ont pas changé par rapport au snapshot précédent.
      - Le problème de la full table ingestion, c'est la charge sur la machine de DB, et l'énorme quantité de données qu'on finit par avoir.
    - Une autre solution peut être l'incremental table ingestion.
      - Il s'agit toujours de récupérer des snapshots à intervalles réguliers, mais seulement de la donnée qui a changé depuis le précédent snapshot.
      - Pour savoir quelle donnée a changé :
        
        La table d'origine doit avoir un champ LAST_MODIFIED, mis à jour automatiquement par la DB.
        
        En retenant le MAX(LAST_MODIFIED) du dernier run d'ingestion (qu'on appelle le highest watermark), on peut construire une query qui récupère uniquement les nouvelles données :
        SELECT * FROM subscriptions WHERE LAST_MODIFIED > "2019-05-01 17:01:00"
        
        On pourra mettre le highest watermark dans le technical metadata layer.
        
        AWS Glue gère nativement le stockage de ce genre de données, mais on peut le mettre dans une DB managée comme Google Cloud SQL ou Azure SQL Database.
      - Cette incremental table ingestion permet d'ingérer moins de données dupliquées, mais elle a encore des inconvénients :
        
        Il faut faire du processing pour faire apparaître les données supprimées, en comparant les snapshots entre eux.
        
        Les données qui sont insérées puis supprimées entre deux snapshots ne seront pas capturées par ce mécanisme, donc on perd des informations.
  - 2 - Le Change Data Capture (CDC).
    - Le CDC permet de récupérer l'ensemble des opérations qui ont lieu sur la table, sans aucun doublon.
    - Il s'agit de lire le log de changements créé par la DB, à l'aide d'une application qui sait le faire.
      - L'application peut être fournie par la DB, ou une application cloud-native comme AWS Database Migration Service, ou une application open source comme Debezium.
    - Etant donné que les DBs ne gardent pas longtemps leur log de changements, le CDC nécessite une infrastructure de type streaming pour être récupéré.
    - Le format des messages récupérés depuis le log de changements contient la valeur du row avant, sa valeur après l'opération, le type d'opération, et des metadata.
      - On va vouloir mettre dans le data warehouse uniquement la valeur après l'opération et le type d'opération.
      - La table dans le data warehouse ressemble du coup au cas de l'incremental table ingestion : on a une entrée par changement.
    - Le CDC sur une DB Oracle.
      - Oracle fournit Oracle GoldenGate, une application qui permet de lire son log de changement et de le transférer vers diverses plateformes.
        
        Il faut acheter la licence pour pouvoir l'utiliser.
      - On peut mettre en place Debezium qui est open source, mais il faudra qu'il puisse se connecter à Oracle XStream API, qui lui-même nécessite quand même une licence GoldenGate.
      - Oracle fournit un outil d'analyse qui s'appelle LogMiner, qui est considéré comme pas 100% fiable.
        
        Certains outils comme AWS Database Migration Service l'utilisent malgré tout.
      - Une alternative moins chère à GoldenGate peut être SharePlex, un produit fait par Quest.
    - Le CDC sur une DB MySQL.
      - MySQL écrit les changements dans un log servant principalement à la réplication, pour ajouter des DBs followers.
      - Vu que c'est une DB open source, il existe de nombreux outils pour servir d'application CDC à partir de ce log, par exemple : Debezium et Apache NiFi.
    - Le CDC sur une DB MS SQL Server.
      - MS SQL Server fournit la possibilité de rediriger le log de changements d'une table vers une table créée spécialement pour ça.
        
        On peut donc facilement implémenter un outil CDC qui n'a qu'à utiliser SQL pour lire cette nouvelle table régulièrement.
      - Parmi les outils qui supportent le CDC sur MS SQL Server, il y a par exemple: Debezium, Apache NiFi et AWS Database Migration Service.
    - Le CDC sur une DB PostgreSQL.
      - PostgreSQL supporte le fait de fournir son log de changements sous un format Protobuf ou JSON, ce qui facilite le travail des applications CDC.
      - Il existe de nombreux outils qui savent lire ces données, par exemple : Debezium et AWS Database Migration Service.
- Concernant le mapping des données depuis la DB vers le data warehouse, il va falloir faire une analyse pour vérifier la compatibilité :
  - 1 - On prépare une liste des types de données supportées par la DB dont on veut capturer les données.
    - Il vaut mieux prendre l'ensemble des types, en prévision d'ajout de colonnes avec des types qui n'étaient pas utilisés jusque là par l'application.
  - 2 - On prépare une liste des types supportés par le data warehouse de destination, et on identifie les différences avec la précédente.
  - 3 - On identifie les types qui ne correspondent pas exactement, mais permetteront une conversion sans perte d'information.
    - Par exemple un type SMALLINT sur MySQL comme source, et le seul entier disponible sur Google BigQuery qui est l'équivalent d'un BIGINT.
  - 4 - On identifie les types qui n'ont pas de correspondance satisfaisante, et pourraient mener à une perte d'information.
    - On essaye de voir si on ne peut pas trouver un workaround, par exemple transformer des données géospatiales en string, puis utiliser du processing pour les parser au moment de la lecture.
  - 5 - Si on est devant une impasse, on essaye de voir s'il n'y a pas un outil de data warehouse plus adapté.
  - 6 - Dans le cas où notre application d'ingestion n'est pas faite à la main, on vérifie les types qu'elle supporte, et leur compatibilité avec la source et la destination.
    - Les auteurs conseillent de faire plusieurs PoC, et disent de ne pas faire confiance aux documentations de ces outils.
  - 7 - Si on écrit l'application d'ingestion à la main, il faut vérifier les types supportés par le driver qui nous permet d'accéder à la DB. Par exemple le driver JDBC.
- Les DBs NoSQL sont à traiter différemment des DBs relationnelles.
  - Parmi les solutions courantes :
    - On peut utiliser un outil SaaS commercial qui supporte notre DB NoSQL : dans ce cas rien de plus à faire.
    - Implémenter l'application d'ingestion à la main, en utilisant l'API de notre DB NoSQL directement pour accéder aux données.
    - Utiliser une application CDC si c'est disponible.
      - Par exemple Debezium supporte MongoDB.
    - On peut utiliser l'outil d'export de données de notre DB NoSQL, et le faire tourner régulièrement pour avoir un snapshot des données.
      - MongoDB permet d'obtenir les données sous un format CSV ou JSON, et l'outil permet d'ajouter des requêtes, donc on peut avoir une colonne qui a la date de la dernière modification, et faire un incremental table ingestion.
      - Cassandra permet d'obtenir les données sous un format CSV, mais uniquement en mode full table ingestion.
- Concernant les metadata liées à l'ingestion.
  - Il faut sauvegarder un certain nombre de statistiques pour pouvoir ensuite faire des vérifications sur la qualité des données ingérées, et du monitoring de l'ingestion.
    - On va mettre tout ça dans notre technical metadata layer.
  - Parmi les statistiques qu'on veut :
    - Le nom et l'adresse IP du serveur de DB.
    - Le nom de la base de données ou du schéma.
    - Le nom de la table.
    - Le type de DB dans le cas où on en gère plusieurs.
    - Pour de l'ingestion en batch, le nombre de rows ingérées.
      - On pourra à partir de ça vérifier que l'ensemble des données sont arrivées à destination.
      - On peut monitorer ce chiffre pour être alerté dans le cas d'une variation anormale.
    - La durée de chaque job d'ingestion, de même que le début et la fin de l'ingestion.
      - C'est un moyen de monitorer la santé de la pipeline.
    - Pour de l'ingestion en streaming, on prend les statistiques par fenêtre temporelle.
      - Par exemple insérer un row toutes les 5 mn dans notre DB de technical metadata. Plus on a besoin de réagir vite, et plus on va choisir une fenêtre petite.
      - On peut aussi ajouter le nombre d'inserts, updates, deletes etc. pour chaque fenêtre.
    - Les changements dans le schéma de la DB source, ce qui nous permettra d'être alerté et d'adapter la pipeline.
Concernant le cas des fichiers.
- Les fichiers (par exemple CSV, JSON) permettent un bon découplage entre deux systèmes.
- On a en général deux moyens de mettre à disposition des fichiers :
  - Via un serveur dédié qui expose un protocole FTP.
  - Via le service de storage d'un cloud provider.
    - Les avantages sont l'aspect elastic, et les mécanismes de sécurité pré-configurés.
    - Le désavantage principal c'est que c'est cloud provider met en place des coûts pour faire sortir la donnée de son infrastructure.
- Les fichiers sont immutables une fois qu'ils sont écrits, ce qu'on aura besoin de tracker c'est quels fichiers ont déjà été ingérés.
  - 1 - Une approche recommandée par les auteurs c'est d'avoir deux dossiers dans le système source qui met à disposition les fichiers : incoming et processed :
    - L'application d'ingestion va ingérer un fichier depuis incoming, puis une fois que l'ingestion est terminée, elle va le copier dans processed et le supprimer d'incoming.
    - On le laisse dans processed pendant quelques jours dans un but de débug et de replay, avant de le supprimer.
    - Parmi les avantages :
      - On n'a pas besoin de tracker quels fichiers ont été traités : il suffit de traiter ceux du dossier incoming.
      - On peut facilement rejouer l'ingestion en replaçant le fichier depuis processed vers incoming.
  - 2 - Dans le cas où l'approche des deux dossiers n'est pas possible, parce que le système source veut organiser autrement ses fichiers, ou qu'on n'a pas la possibilité de les modifier, on peut mettre en place l'approche des timestamps.
    - Chaque fichier va avoir un timestamp de la dernière fois qu'il a été modifié, et on va devoir garder le timestamp le plus récent dont on a ingéré un fichier dans le technical metadata layer.
    - Vu que le filesystem ne fournit pas de système d'indexation, on va devoir lire à chaque fois les metadata de l'ensemble des fichiers pour savoir s'ils sont plus récents ou moins récents que notre timestamp sauvegardé.
      - On peut se retrouver face à un problème de performance, surtout avec les stockages cloud de masse.
    - Cette méthode rend plus compliqué le replay des fichiers : on devra possiblement modifier notre dernier timestamp sauvegardé, et on aura du mal avec les fichiers qui ont le même timestamp de modification.
    - Certains outils comme Apache NiFi implémentent déjà ce mécanisme.
      - Il faudra faire attention à faire un backup de ces données pour ne pas avoir à reprocesser tous les fichiers.
  - 3 - Une variante de l'approche des timestamps consiste à organiser les fichiers source dans une arborescence de dossiers représentant la date d'ajout.
    - Exemple :
      /ftp/inventory_data/incoming/2019/05/28/sales_1_081232
    - On peut s'en servir pour ne lire que les metadata des fichiers qui sont à la date qu'on veut ingérer.
  - 4 - Des outils cloud-native existent pour copier des fichiers d'un storage à un autre, et de ne copier que les nouveaux fichiers à chaque fois qu'il y en a.
    - gsutil permet de le faire chez Google Cloud, blobxfer chez Azure, et s3 sync chez AWS.
    - Il est compliqué de faire du replay avec ces outils, parce qu'il n'y a pas de dernier timestamp stocké à modifier.
- Concernant les metadata techniques à garder :
  - On ne va pas à ce stade récupérer de statistiques sur le nombre de rows dans le fichier, parce que ce serait techniquement coûteux pour l'ingestion layer.
    - On le fait pour les DBs parce que c'est pas cher.
  - Parmi les statistiques à récupérer :
    - Nom permettant d'identifier la source.
    - Taille du fichier.
    - Durée de l'ingestion.
    - Le nom du fichier et le path où il était (peut contenir des infos importantes).
Concernant le cas des streams.
- Il s'agit ici de lire de la donnée disponible dans Kafka, ou encore dans un équivalent cloud-native.
- On parle ici seulement d'ingestion en mode streaming, c'est-à-dire que la donnée est disponible dans la plateforme dès que possible, mais elle sera exploitée plus tard.
- Les étapes à mettre en place sont :
  - 1 - La 1ère étape est de lire le stream source, et de l'écrire dans le fast storage de notre cloud data platform, qui est aussi un stream.
    - On peut faire ça avec Kafka Connect, qui permet de lire et écrire entre deux topics Kafka, mais aussi de lire depuis Kafka et écrire dans une solution de streaming cloud-native, ou l'inverse.
    - On peut aussi faire notre propre application consumer Kafka à la main, mais il faudra alors s'occuper de la gestion des erreurs, du logging, et du scaling de notre consumer. Les auteurs le déconseillent.
  - 2 - On va ensuite l'écrire dans le data warehouse.
    - Les auteurs conseillent fortement d'utiliser une solution cloud-native pour ça, en fonction de notre fast storage :
      - Azure Stream Analytics qui lit depuis Azure Event Hubs pour écrire dans Azure SQL Warehouse.
      - Google Cloud Dataflow qui lit depuis Cloud Pub/Sub pour écrire dans BigQuery.
      - AWS Kinesis Data Firehose qui lit depuis AWS Kinesis pour écrire dans Redshift.
    - Pour BigQuery on peut ingérer dans le data warehouse en streaming, mais pour les deux autres, il faudra faire de petits batchs.
  - 3 - L'autre chose à faire en parallèle c'est d'écrire la donnée depuis le fast storage vers le slow storage.
    - On peut là aussi utiliser les solutions cloud-natives.
    - Il va falloir écrire la donnée par batchs pour des raisons de performance. Les auteurs recommandent des batchs de plusieurs centaines de MB si c'est possible.
- Kafka (et les solutions cloud-natives similaires) doit faire le commit de son offset, et en général il le fait après avoir traité plusieurs messages pour des raisons de performance;
  - Ça veut dire que si il y a un crash, les message traités mais non commités seront traités à nouveau. Donc il faut gérer la duplication.
  - Un des moyens de le faire c'est d'avoir un identifiant unique par message, et ensuite d'enlever les doublons dans la phase de processing.
- Kafka a en général une cleanup policy qui est de l'ordre de la semaine, ce qui fait que pour rejouer de la donnée, il faut prévoir une étape qui va la chercher dans le slow storage, et la remet dans le fast storage.
- Concernant les metadata techniques à garder :
  - Les metadata à garder ressemblent à ceux du cas CDC depuis les DBs relationnelles.
  - On mesure le nombre de messages ingérés par fenêtre de temps (dont la taille dépendra du type de données ingérées).
Concernant le cas des applications SaaS qui fournissent de la donnée.
- Les applications SaaS vont en général exposer leurs données via une API REST, le contenu étant formaté en JSON ou parfois en XML.
  - Il faut d'abord s'authentifier, souvent avec OAuth.
  - Et ensuite il faut étudier la documentation du provider SaaS pour savoir quel call faire.
- Il y a un certain nombre de difficultés.
  - Chaque provider va designer son API selon ses contraintes. Et donc si on veut supporter de nombreux providers, il va falloir adapter l'ingestion layer pour chacun d'entre eux.
  - Chaque provider va fournir soit du full data export soit de l'incremental data export, et parfois les deux.
    - Le full data export consiste à obtenir une liste d'objets, puis à aller chercher les données pour chacun d'entre eux.
    - L'incremental data export consiste à obtenir une liste d'objets qui ont changé entre deux timestamps qu'on fournit, pour ensuite aller chercher leurs données récentes uniquement.
  - Le JSON reçu est en général imbriqué sur plusieurs niveaux.
    - Certains data warehouses gèrent les données imbriquées, mais ce n'est pas le cas de Redshift pour lequel il faudra faire une étape de processing pour mettre ces données à plat.
    - De manière générale, mettre les données à plat dans plusieurs tables plus petites est plus pratique pour les data scientists.
- Etant donné la difficulté à implémenter et maintenir une pipeline ingérant de la donnée de sources SaaS, les auteurs conseillent de bien réfléchir à l'implémenter soi-même.
  - S'il s'agit d'une source pas trop compliquée, ça peut passer.
  - Si par contre il s'agit de nombreuses sources, alors il nous faudra une grande quantité de code et de maintenance.
    - Les auteurs conseillent plutôt une solution off-the-shelf comme Fivetran qui supporte la plupart des sources SaaS connues.
- Concernant les metadata techniques à garder :
  - Il s'agit du même type de metadata que pour les sources en batch comme les DBs ou les fichiers.
  - On voudra notamment :
    - Le nom de la source.
    - Le nom de l'objet qu'on va chercher dans la source.
    - Les temps de début et fin d'ingestion.
    - Le nombre de rows qu'on récupère.
Pour des questions de sécurité, il est préférable d'encapsuler notre cloud data platform dans un virtual private cloud (VPC).
- Pour faire le lien entre la plateforme dans le VPC et la donnée qu'on veut aller chercher, on peut utiliser un VPN Gateway, qui permet de passer par internet de manière sécurisée.
- Dans le cas des SaaS comme source, ils fournissent des APIs sécurisées par HTTPS, et disponibles globalement sur internet, donc il n'est pas nécessaire d'établir une connexion via VPN Gateway.
- Dans le cas où on veut transférer des centaines de GB par jour, il vaut mieux mettre en place une connexion directe.
  - Les solutions cloud-natives ont leur outil de connexion directe : AWS Direct Connect, Azure ExpressRoute, Google Cloud Interconnect.

5 - Organizing and processing data

Les architectes de l'ancienne école ont encore tendance à recommander de faire le processing dans le data warehouse.
- Les auteurs du livre suggèrent que la manière moderne est de le faire sur des machines à part, par exemple avec Spark, qui lirait et écrirait dans le data lake.
- Les arguments sont les suivants ((1) pour faire le calcul dans le data warehouse, et (2) pour utiliser la layered architecture) :
  - Flexibility : avec la (1) le résultat du processing n'est utilisable que dans le data warehouse, avec la (2) on peut facilement le rediriger ailleurs.
  - Developer productivity : il y a plus de personnes qui connaissent le SQL, donc le (1) a un avantage court terme, bien que Spark soit plus puissant, il faut souvent former les devs.
  - Data governance : la source principale étant le data lake, faire les transformations au même endroit permet d'être sûr d'avoir toutes les versions alignées. Dans le cas où on fait ça dans le data warehouse, il est préférable de ne pas le faire dans le data lake pour ne pas avoir de divergence.
  - Cross-platform portability : changer de cloud vendor est bien plus simple avec Spark qu'avec du code SQL qu'il faudra au moins tester.
  - Performance : avec la (1) le processing impacte le data warehouse, avec le (2) on fait le calcul complètement à part et on n'impacte personne.
  - Speed of processing : avec la (1) on peut faire du real time analytics dans certains cas avec difficulté, avec la (2) ça marche facilement.
  - Cost : tous les providers de data warehouse ne le font pas (mais ils vont finir par le faire), mais pour ceux qui font payer le processing ça revient plus cher que de faire le processing sur des machines complètement à part.
  - Reusability : avec la (1) on peut parfois utiliser des stored procedures, avec la (2) on a du code qu'on peut directement réutiliser.
Le processing se décompose en stages.
- Chaque stage contient : une area de stockage dans le data lake, et un job de calcul distribué (par exemple avec Spark), qui va créer la donnée pour l'étape suivante.
  - Les jobs sont coordonnés par l'orchestration layer.
  - Les jobs peuvent être de deux types :
    - Common data processing : les transformations communes, par exemple dédupliquer les messages, valider les dates etc.
    - Business logic specific processing : les transformations spécifiques à chaque use-case, qui vont par exemple filtrer les campagnes de marketing à succès uniquement si le use-case c'est d'afficher les meilleures campagnes.
- Avoir un ensemble de stages standardisés est important pour que chacun puisse s'y retrouver malgré le scale.
- Les étapes proposés par les auteurs sont :
  - 1 - Landing area : c'est là que la donnée arrive en premier, il ne s'agit pas d'un stockage long terme.
  - 2 - Staging area : la donnée subit des checks basiques de qualité, et on vérifie qu'elle est conforme au schéma attendu. Elle est stockée sous format Avro.
  - 3 - Archive area : la donnée est copiée depuis la landing area vers l'archive area.
    - Cette opération n'est effectuée qu'après que la donnée ait pu aller vers la staging area avec succès.
    - On pourra refaire le processing de la donnée simplement en la copiant depuis l'archive area vers la landing area.
  - 4 - Production area : la donnée subit la transformation business nécessaire pour un use-case particulier avant d'aller là.
    - Elle est aussi transformée du format Avro vers Parquet, qui est plus adapté pour faire de l'analytics.
    - 4.1 - Pass-through job : il s'agit d'un job qui copie la donnée de la staging area vers la production area sans transformation autre que le format Parquet, et ensuite la copie dans le data warehouse.
      - Ce use-case “basique” est utile pour débugguer les autres use-cases.
    - 4.2 - Cloud data warehouse and production area : les use-cases qui ont besoin de la donnée dans le data warehouse passent d'abord par le processing de la staging area vers la production area.
  - 5 - Failed area : chaque étape peut faire face à des erreurs, qu'elles soient liées à la donnée ou à des échecs temporaires de la pipeline.
    - Les messages qui n'ont pas réussi une étape vont dans cette area où on pourra les examiner et voir ce qu'il faut corriger.
    - Une fois la correction faite, il suffit de les copier dans l'area de l'étape où ils ont échoués.
- Chaque area doit être dans un container du service de stockage de notre cloud provider.
  - Les containers contiennent des folders.
  - Ils sont appelés buckets chez AWS et GCP.
  - C'est au niveau de ces containers qu'on peut configurer les droits d'accès, et choisir le prix qu'on paye pour les performances qu'on aura (hot / cold / archive storage).
    - Parmi nos 5 areas, toutes sont de type hot, sauf l'archive area qui peut être cold / archive.
On a besoin d'une organisation des folders claire dans chaque area.
- Les éléments communs sont :
  - Le namespace représente la catégorisation la plus high level, pour les petites organisations ça peut être juste le nom de l'organisation, mais pour les plus grandes ça peut être le nom du département.
  - Le pipeline name représente le nom d'un job en particulier. Il faut qu'il soit clair par rapport à ce que fait le job, et utilisé partout pour parler de lui.
  - Le data source name identifie une source. C'est l'ingestion layer qui choisit ce nom et le note dans le metadata layer.
  - Le batchId représente l'identifiant de chaque batch de donnée écrit dans la landing area par l'ingestion layer.
    - On peut utiliser un UUID pour le représenter, ou encore un ULID, qui a la particularité d'être plus court et de permettre de savoir facilement si un autre ULID est plus grand ou plus petit.
- Pour la landing area, les auteurs proposent la folder structure :
  - landing/NAMESPACE/PIPELINE/SOURCE_NAME/BATCH_ID/
    - landing représente le nom du container.
  - Exemple : /landing/my_company/sales_oracle_ingest/customers/01DFTQ028FX89YDFAXREPJTR94/
- Pour la staging area, de même que pour les autres areas, il s'agit de stocker la donnée sur le long terme, donc on aimerait une structure qui fasse apparaître le temps, avec 3 folders supplémentaires :
  - Il s'agit d'ajouter 3 folders supplémentaires qui viennent de la convention de Hadoop : year=YYYY/month=MM/day=DD.
  - Exemple : /staging/my_company/sales_oracle_ingest/customers/year=2019/month=07/day=03/01DFTQ028FX89YDFAXREPJTR94/
  - De nombreux outils (y compris Spark) vont reconnaître ce format, et si notre batchId est un ULID, les folders les plus récents seront présentés en premier.
- Pour la production area, on ne peut pas vraiment reporter les sources qui ont servi à la donnée dans le folder name - il y en a potentiellement des dizaines.
  - On va donc plutôt créer des sources dérivées dont on mettra le nom à la place de la source, et on documentera ces sources dérivées dans le metadata layer.
La donnée qui arrive en streaming :
- Va passer directement vers la version streaming du processing layer sans être stockée d'abord dans le slow storage. C'est traité au chapitre 6.
- Mais on va quand même l'envoyer dans le slow storage en parallèle pour un but d'archivage et rejeu si besoin.
  - Un job va lire depuis le fast storage où arrive la donnée en streaming, par batchs suffisamment gros, et va écrire ça dans la landing area.
  - Les fichiers seront ensuite passés de stage en stage jusqu'à la production area.
Parmi les common processing steps :
- File format conversion.
  - L'approche data lake traditionnelle consiste à laisser les données telles quelles, et laisser chaque pipeline parser elle-même la donnée et faire les traitements dont elle a besoin.
    - Mais cette approche a du mal à scaler.
    - Dans la cloud data platform architecture, on choisit de faire certains traitements en amont, pour éviter d'avoir à tester et maintenir du code qui fait ça dans chaque pipeline.
  - Avro et Parquet sont des formats binaires intégrant un schéma.
    - Ils permettent de ne pas répéter le nom des champs, et donc d'économiser de la place.
    - Ils permettent de garantir le schéma de la donnée.
    - Avro est organisé en blocs de rows, alors que Parquet est organisé en blocs de columns.
      - Les fichiers organisés en rows sont utiles quand on lit la donnée de toutes les colonnes pour certains rows donnés. La staging area sert principalement à faire des transformations ou de l'exploration ad-hoc, donc Avro est adapté.
      - Les fichiers organisés en columns sont utiles quand on ne veut traiter qu'une column sur un grand nombre de rows. La production area sert à faire des requêtes d'analytics, donc Parquet est adapté.
  - Pour la conversion depuis le format initial vers Avro, puis vers Parquet, Spark permet de lire et écrire ces différents formats.
    - Exemple :
      clicks_df = spark.read.json(in_path) clicks_df = spark.write.format("avro").save(out_path)
- Data deduplication.
  - On s'intéresse ici au fait d'avoir un attribut sur notre donnée qui soit unique dans l'ensemble des données.
    - A partir du moment où on n'a pas de garanties d'unicité, on peut se retrouver dans une situation de duplication, par exemple si le metadata repository est corrompu, si la source envoie une donnée dupliquée, ou encore si un dev rejoue certaines données qui avaient déjà marché.
    - Le problème existe aussi avec Kafka, où des transactions existent si on lit un record et qu'on écrit dans un topic Kafka, mais pas si on écrit sur un service de storage.
  - Spark a une fonction intégrée dropDuplicates() qui permet de dédupliquer en fonction d'une ou plusieurs colonnes.
    - On peut dédupliquer sur un batch qui arrive dans la landing area pour pas cher :
      users_df = spark.read.format("csv").load(in_path) users_deduplicate_df = users_df.dropDuplicates(["user_id"])
    - Si on veut vraiment dédupliquer sérieusement, il faut aussi joindre l'ensemble des données déjà présentes dans la staging area au batch courant, et appliquer la déduplication dessus, par exemple avec du SQL qu'on passe à Spark.
      incoming_users_df .createOrReplaceTempView("incomgin_users") staging.users_df .createOrReplaceTempView("staging_users") users_deduplicate_df = spark.sql( "SELECT * FROM incoming_users u1 LEFT JOIN staging_users u2 ON u1.user_id = u2.user_id WHERE u2.user_id IS NULL" )
    - Le problème c'est que la déduplication à chaque fois avec l'ensemble des données coûte cher. Donc il faut vérifier que notre use-case le nécessite d'un point de vue business.
      - On peut aussi dédupliquer avec seulement les fichiers dans le dossier de l'année actuelle, du mois actuel etc. depuis la staging area.
- Data quality checks.
  - Une vérification minimale de la qualité de la donnée est en général nécessaire pour la plupart des cas d'usages. Par exemple :
    - La longueur de certaines colonnes.
    - La valeur numérique acceptable de certaines colonnes.
    - Le fait d'avoir certaines colonnes “obligatoires”.
    - Le fait d'avoir certaines colonnes respecter un pattern, par exemple l'email.
  - Spark a la fonction filter() qui permet d'obtenir les colonnes qui respectent une mauvaise condition.
    - On a aussi subtract() qui permet d'enlever ces rows du batch, pour passer les rows valides à la production area, et les rows invalides à la failed area.
      - Attention à la consistance des données, en fonction du contexte business, il peut être plus judicieux de laisser passer la donnée, et de simplement informer les data engineers du problème.
      - De manière générale, il faut réfléchir à la criticité de chaque problème de qualité pour décider quoi faire en cas de donnée malformée : filtrer la donnée, laisser passer et prévenir quelqu'un, ou annuler l'ingestion du batch entier.
    - Exemple :
      users_df = spark.read.format("csv").load(in_path) bad_user_rows = users_df.filter( "length(email) > 100 OR username IS NULL" ) users_df = users_df.subtract(bad_user_rows)
On peut créer des jobs configurables : l'orchestration layer lance un job, en lui donnant d'abord la configuration contenant les sources à traiter, le schéma à valider en fonction des sources, la folder structure où insérer les nouveaux fichiers etc.
- Ça permet d'économiser du code, au moins pour les jobs de transformation “common”.
- Le bon endroit pour la configuration c'est le metadata layer.
- Pour déclencher nos jobs, il faut qu'il y ait une forme de monitoring de la landing area, soit avec du code qu'on écrit nous-mêmes, soit avec la fonctionnalité de monitoring d'un outil d'orchestration cloud.

6 - Real-time data processing and analytics

La notion de real-time (ou streaming) dans le contexte d'une pipeline data peut recouvrir deux choses différentes :
- 1 - real-time ingestion : on ingère la donnée une par une avec un mécanisme de message streaming, et on l'amène jusqu'au data warehouse. Mais la consommation de la donnée ne se fait pas en temps réel.
  - L'aspect “real-time” ne concerne que l'ingestion layer.
  - Le processing se fait à la demande, et peut prendre des secondes voire des minutes, mais il se fait sur une donnée fraîche.
  - Il peut se faire selon un schedule, ou à la demande des utilisateurs humains qui attendront un peu avant d'avoir un résultat.
  - Exemple : un data analyste veut pouvoir exécuter une requête pour afficher un dashboard sur des données fraîches quand il en a besoin. Le dashboard n'est pas mis à jour en continu mais juste à l'exécution de cette requête.
- 2 - real-time processing : on récupère la donnée une par une, et on la redirige vers un autre système qui va réagir à chaque donnée qui arrive pour se mettre à jour.
  - Le real-time processing nécessite la real-time ingestion. L'aspect “real-time” concerne donc l'ingestion layer et le processing layer.
  - On est dans un cas d'usage où on a besoin que le processing se fasse très vite et en continu, en général à destination d'un autre système.
  - Exemple : la donnée qui arrive dans la pipeline est ensuite mise à disposition d'un système de jeu vidéo pour adapter le comportement du jeu en fonction de ce que fait le joueur en temps réel. Par exemple, ajuster la probabilité de faire apparaître un monstre.
  - La donnée est traitée par un real-time job qui tourne en permanence et ajuste les calculs en fonction des nouvelles données.
    - Elle est ensuite mise à disposition d'un key/value store ou éventuellement d'une DB relationnelle, pour un accès rapide. Le data warehouse est trop lent et est fait pour des requêtes à la demande sur de grandes quantités de données.
    - Elle peut aussi être postée dans le fast storage, c'est-à-dire comme event de streaming pour déclencher un autre processing.
    - C'est parce que ce job tourne en continu avec des choses chargées en RAM qu'il donne un résultat rapide, contrairement à une requête SQL dans un data warehouse par exemple, qui ne se déclenche qu'au moment où on la lance.
- Il est très important de clarifier le besoin : dans le cas où on n'a besoin que de real-time ingestion, la complexité de mise en œuvre est beaucoup moins grande.
  - Les auteurs conseillent d'utiliser la real-time ingestion plutôt que la batch ingestion, sauf quand la source ne supporte pas le real time.
    - La real-time ingestion implique moins de nécessité d'orchestration et de monitoring.
    - Pour éviter l'incohérence pour les utilisateurs, il vaut mieux éviter de mixer des données real-time avec des données qui viennent en batch.
    - On n'a en général pas la possibilité d'utiliser le même système pour traiter les données qui arrivent en real-time et les données qui arrivent par batch.
      - Par exemple Google Cloud Dataflow le permet avec l'utilisation de Beam, mais la plupart du temps on aura besoin de deux outils.
  - Selon les auteurs, la plupart du temps quand les utilisateurs demandent du “real-time”, ils veulent en fait juste de la real-time ingestion.
    - Quand des utilisateurs demandent à pouvoir afficher leur dashboard en “real-time” alors qu'il tourne une fois par jour, bien souvent avoir de la real-time ingestion et faire tourner le processing du rapport toutes les heures ou toutes les 15 minutes leur suffira.
    - Parmi les cas d'usage qui pourraient nécessiter du real-time processing : les systèmes d'action in-game, les systèmes de recommandation, les systèmes de détection de fraude.
  - Transiter une pipeline de la batch ingestion à la real-time ingestion se fait sans trop de difficulté.
    - Transiter du batch processing au real-time processing est bien plus complexe vu qu'il va falloir en général changer d'outils, et donc il faut penser ça en amont.
Le fast storage est composé d'un système d'event streaming du type Kafka.
- Les messages (qui font entre quelques KB et 1 MB) sont traités un par un, et stockés dans des topics. Ils sont identifiables par leur offset.
- Les producers écrivent dans les topics, et les consumers lisent depuis les topics. Kafka a des mécanismes qui leur permettent de publier et consommer de manière fiable malgré les fautes.
- Pour permettre de scaler, le contenu des topics est séparé en partitions, qui se trouvent sur des machines différentes, avec des copies pour plus de fiabilité.
- Là où lire et écrire dans S3 mettrait quelques centaines de ms, le faire dans Kafka en prend 10 fois moins, mais surtout Kafka tient la charge avec une très grande quantité de petits messages.
De même que pour le slow storage et le batch processing, le fast storage est organisé en areas qui servent à des stages de processing.
- Les étapes sont :
  - 1 - Landing area : l'ingestion layer écrit la donnée dans cet endroit.
  - 2 - Staging area : la donnée subit des checks basiques de qualité, et on vérifie qu'elle est conforme au schéma attendu.
  - 3 - Archive area : la donnée est copiée depuis la landing area vers l'archive area.
    - Il s'agit d'espace de stockage cloud classique.
    - On pourra refaire le processing de la donnée simplement en la copiant depuis l'archive area vers la landing area.
  - 4 - Production area : la donnée subit la transformation business nécessaire pour un use-case particulier avant d'aller là.
    - 4.1 - Pass-through job : il s'agit d'un job qui copie la donnée de la staging area vers la production area sans transformation, et ensuite la copie dans le data warehouse.
      - Ce use-case “basique” est utile pour débugguer les autres use-cases.
    - 4.2 - Staging to production : des jobs lisent la donnée à partir de la staging area dans un but de reporting/analytics, et créent un dataset dans la production area, pour charger la donnée ensuite dans le data warehouse ou dans une DB relationnelle ou NoSQL.
  - 5 - Failed area : chaque étape peut faire face à des erreurs, qu'elles soient liées à la donnée ou à des échecs temporaires de la pipeline.
    - Les messages qui n'ont pas réussi une étape vont dans cette area où on pourra les examiner et voir ce qu'il faut corriger.
    - Une fois la correction faite, il suffit de les copier dans l'area de l'étape où ils ont échoués.
- Côté organisation en topics :
  - Les providers limitent en général le nombre de topics à quelques milliers, et c'est l'abstraction principale qu'on a. Avec des centaines de tables par DB qu'on utilise comme source, les topics sont vite très nombreux.
  - Selon les auteurs, l'organisation la plus pertinente pour le cas général serait d'utiliser un topic par area, et de faire la distinction entre sources avec un champ à l'intérieur des messages.
    - Mais dans le cas où on a des sources qui donnent des messages structurés très différemment, ou qui ne permettent pas d'utiliser des jobs de processings communs, on peut faire des topics différents par source.
    - Une autre raison de séparer en topics par source peut être la limitation en termes de quotas par topic, de la part du provider.
    - Une autre raison pour publier dans des topics différents peut être la structure interne des équipes, et les questions de sécurité, pour restreindre certaines données à certaines équipes.
- Pour rendre les jobs configurables, on peut faire en sorte qu'ils lisent le contenu du message et appellent une librairie qui va faire quelque chose de particulier en fonction de la valeur lue.
Parmi les transformations qu'on a couramment dans les systèmes real-time, il y a la déduplication des messages.
- Les duplications sont courantes dans les systèmes real-time, elles ont deux origines :
  - 1 - Des duplications issues de la source, sur lesquelles on n'a pas de contrôle.
  - 2 - Des duplications qui sont dues au fonctionnement des systèmes real-time, et à leur nature distribuée.
    - On peut par exemple avoir un producer qui envoie un message, mais ne reçoit pas l'acknowledgement à cause d'un problème réseau. Un autre broker sera élu master de la partition et on se retrouvera avec une duplication.
    - Côté consumer, il suffit que l'un d'entre eux envoie un message et crash avant de commiter. Il va alors renvoyer le même message quand il reviendra à la vie.
- La difficulté pour dédupliquer avec les systèmes real-time c'est qu'on a une donnée qui arrive en permanence, et qui est distribuée sur plusieurs machines.
  - Une solution peut être d'utiliser une time window : on choisit un début et une fin de timestamp, et on récupère tous les messages correspondants pour faire une déduplication parmi eux.
    - On peut avoir par exemple une sliding window qui se déplace dans le temps, ou tumbling window qui va diviser le temps en tranches disjointes.
    - Le problème c'est qu'on est limités sur la tranche de temps qu'une machine peut traiter, et la déduplication ne se fait que pour les messages de cette tranche, et pas avec les autres tranches.
  - Une autre solution est d'avoir un key/value cache dans lequel on met l'ID de chaque message traité, et qu'on réinterroge à chaque fois pour éviter de le retraiter encore.
    - La taille va rarement être un problème : stocker 1 milliard de UUID fait ~15 Go.
    - Par contre il faut que le store soit highly available et performant, donc une solution cloud est bien adaptée.
    - Exemples de key/value stores : Azure Cosmos DB, Google Cloud Bigtable, AWS DynamoDB.
  - Une 3ème solution peut être de laisser les messages dupliqués jusqu'au data warehouse, et dédupliquer ensuite par un job en mode batch, soit dans le data lake, soit dans le data warehouse.
    - Ca permet d'avoir une real-time ingestion particulièrement rapide, mais il faut que la duplication soit OK dans un premier temps.
Une autre transformation courante est la conversion de format.
- Les messages dans le système real-time sont consommés un par un, donc il est capital d'avoir des schémas bien définis entre producers et consumers. Le metadata layer pourra nous aider à le stocker.
- Concernant le format :
  - JSON ne fournit pas de mécanisme de gestion de schéma, et est plus volumineux. Il peut être compressé, mais ce serait surtout efficace avec plusieurs messages où des noms de champ se répètent par exemple.
  - Avro permet de minimiser la taille du message, et permet une gestion du schéma avec la possibilité de le stocker dans un store.
  - Parquet n'apporte aucun avantage dans un système real-time puisque son but est de permettre de lire de grandes quantités de données pour faire du processing dessus, et qu'on est ici sur du message par message.
Concernant les quality checks, on peut avoir un job qui vérifie la qualité de chaque message avant de le placer dans l'area du stage suivant ou dans la failed area.
- Dans le cas où on a de nombreuses sources gérées par plusieurs équipes, la difficulté va surtout être dans la définition de ce qu'est une donnée avec une qualité suffisante.
- Nos quality checks peuvent impliquer de vérifier une caractéristique impliquant plusieurs messages, par exemple “pas plus de 10% des commandes avec le statut cancelled”.
  - Il faudra alors utiliser les techniques de windowing comme avec la déduplication.
  - Si la durée sur laquelle on veut faire les checks est trop grande par rapport à ce que supportent nos outils, il faudra faire passer le flow par le batch processing.
Dans le cas où on veut combiner une source de données real-time et une source batch, on peut :
- 1 - Avoir le job real-time qui lit le message batch à combiner avec les données du message real-time, et qui le stocke dans sa RAM.
- 2 - Puis ce job combine les deux pour les mettre dans la real-time production area.
- 3 - Et il continue avec les messages suivants en utilisant le message batch qui est dans sa RAM, jusqu'à ce qu'il y en ait un nouveau.
- La limitation pourrait être la taille du message batch : s'il ne rentre pas dans la ram des VMs qui font le real-time processing, on peut fallback sur du batch processing.
Les 3 cloud vendors principaux fournissent chacun deux outils pour le real-time processing : un outil de real-time storage type Kafka, et un outil qui fait le real-time processing.
- AWS.
  - Kinesis Data Streams est équivalent à Kafka.
    - Il fournit des clients dans 5 langages, dont Node.js.
    - Il a l'équivalent des topics sous le nom de Data Streams.
    - Il a l'équivalent des partitions sous le nom de shard, et limite la throughput à 1 MB/s par shard.
    - Il supporte le “resharding” à la hausse ou à la baisse.
    - Il limite la taille des records à 1 MB.
    - La rétention par défaut est d'un jour, et va jusqu'à une semaine.
  - Kinesis Data Analytics est l'outil de processing real-time.
    - Il fournit une API SQL pour créer les jobs, mais c'est limité à des records qui contiendront du CSV ou du JSON.
    - Il fournit aussi une API Java, qui utilise Apache Flink et permet plus de flexibilité sur le format des records.
    - Il ne fournit pas de mécanisme de déduplication.
- GCP.
  - Cloud Pub/Sub est un peu différent de Kafka et il abstrait plus de choses.
    - Il fournit des clients dans 7 langages, dont Node.js.
    - Les topics permettent de regrouper les records, mais il n'y a pas de notion de partition, ou en tout cas elle est abstraite derrière l'API.
    - Les consumers peuvent faire une subscription à un topic pour consommer les records.
      - Ils peuvent aussi utiliser une subscription pour recevoir de la donnée combinée de plusieurs topics.
      - On se sert aussi des subscriptions pour scaler le throughput : on a le droit à 1 MB/s par subscription.
    - Les records sont limités à 10 MB.
    - La rétention des données maximale est d'une semaine.
    - Il ne fournit pas d'offsets pour les records, ce qui limite la possibilité de rejouer certains messages particuliers.
      - On peut faire des snapshots pour pouvoir les rejouer, mais ils sont limités à 5000 par projet.
      - On a aussi la possibilité de rejouer par timestamp, mais c'est peu précis.
  - Cloud Dataflow est l'outil de processing real-time.
    - Il fournit une API SQL pour créer les jobs, mais c'est limité à des records qui contiendront du JSON.
    - Il fournit aussi une API Java et Python, qui utilise Apache Beam et permet plus de flexibilité sur le format des records.
    - Il permet de dédupliquer les messages issus de problèmes techniques, et propose aussi une déduplication des messages par ID, sur une fenêtre de 10 minutes.
- Azure.
  - Event Hubs est équivalent à Kafka.
    - Il fournit des clients en .NET et Python, mais des versions open source sont disponibles pour d'autres langages.
    - Il supporte 3 protocoles pour s'y intégrer en tant que producer ou consumer : HTTPS, AMQP et Kafka. Ça permet de migrer vers Azure sans avoir à tout réécrire.
    - Il a l'équivalent des topics dans le cas de Kafka, ou des hubs dans le cas d'AMQP.
    - Il a l'équivalent des partitions, qu'il faut définir à l'avance comme pour Kafka, et à l'inverse de Kinesis Data Streams pour lequel on peut “resharder”.
    - Le throughput est limité à 1 MB/s ou 1000 messages/s par partition.
    - Les records ne peuvent pas dépasser 1 MB.
    - La période de rétention maximale est d'une semaine.
    - Contrairement à Kinesis Data Streams qui stocke les offsets des consumers dans DynamoDB, ou à Kafka qui le stocke dans un topic interne, Event Hubs laisse cette responsabilité aux consumers.
  - Azure Stream Analytics est l'outil de processing real-time.
    - Il ne propose qu'une API SQL, avec des fonctionnalités avancées de type windowing, recherche dans des dictionnaires etc.
    - Si on veut plus de flexibilité, on peut utiliser Spark à travers Azure Databricks, mais il s'agira de micro-batching et non pas de vrai streaming.
    - Il ne fournit pas de fonctionnalités de déduplication.

7 - Metadata layer architecture

Il existe deux types de metadata dans le cadre de la data.
- 1 - La business metadata permet de donner du contexte à la donnée.
  - Ca peut être par exemple : la source, le propriétaire de la donnée, la date de sa création, la taille de la donnée, le but de la donnée, son niveau de qualité etc.
  - Ça aide notamment à trouver la donnée qu'on cherche.
  - On appelle souvent l'outillage autour de la business metadata le data catalog.
    - Les cloud vendors ont chacun leur outil : Google Cloud Data Catalog, Azure Data Catalog, AWS Glue Data Catalog.
- 2 - La data platform metadata (ou pipeline metadata) permet de rassembler des informations sur les pipelines de données.
  - Ça peut être des informations sur les sources, sur le succès ou l'échec de runs de pipelines, les erreurs qui ont eu lieu etc.
  - Ça permet notamment le monitoring et la configuration des pipelines.
  - Cette metadata est plus alignée avec la responsabilité des data engineers, et c'est sur elle que se concentre ce livre.
Une seule pipeline simple peut être gérée avec du code, mais dès que le système de pipelines se complexifie, il faut gérer cette complexité.
- On a le choix de :
  - 1 - Dupliquer le code des pipelines pour les rendre simples, mais alors il faudra refaire des modifications à plusieurs endroits à chaque fois qu'on voudra changer quelque chose qui concerne plusieurs pipelines.
  - 2 - Mettre du code en commun pour éviter de réécrire trop de choses, mais alors la codebase se complexifie, et l'investigation des problèmes aussi.
- Les auteurs du livre conseillent de mettre le code en commun, et de rendre les pipelines configurables pour éviter l'explosion de complexité.
  - On pourra par exemple mettre en commun l'ingestion de sources de type RDBMS, et celles de type file. Ou encore mettre en commun des jobs de data quality check.
  - Si la configuration se trouve dans un endroit séparé, il devient facile de la changer sans avoir à toucher au code.
  - Parmi les éléments de configuration, il peut y avoir par exemple : l'endroit d'où on récupère la donnée, l'endroit où on l'envoie, les checks de qualité et transformations qu'il faut faire sur chaque donnée etc.
La data platform metadata a 3 fonctions :
- 1 - Stocker les configurations des pipelines.
  - Par exemple, si un path d'input sur un serveur FTP change, il suffira d'aller changer la configuration de la pipeline dans le metadata layer, sans toucher au code.
  - Pour connaître les inputs et outputs d'une pipeline, il suffira aussi de regarder sa configuration.
- 2 - Monitorer l'exécution et le statut des pipelines.
  - Par exemple, en cas d'erreur sur un pipeline, il suffira d'aller regarder dans le metadata layer pour avoir un statut détaillé de la pipeline, avec des statistiques d'échec, de nombre de duplicatas etc.
- 3 - Servir de schema repository.
  - Cette partie sera plus développée dans le chapitre 8.
Il n'existe pas vraiment de standard concernant le modèle d'un metadata layer.
- Les auteurs du livre en proposent un centré autour de 4 domaines, contenant les aspects qu'ils pensent être suffisamment universels.
  - 1 - La Pipeline Metadata contient les informations d'input, output et transformations de chaque pipeline.
    - L'objet Namespace se trouve au plus haut niveau, et permet de séparer des groupes de pipelines.
      - Il s'agit par exemple de pouvoir appliquer des droits d'accès différents à des ensembles de pipelines.
      - On pourra l'utiliser pour nommer les folders, ou les topics de notre système de slow et fast storage.
      - Sa structure est :
        
        ID
        
        Name
        
        Description
        
        Created At
        
        Updated At
    - L'objet Pipeline décrit un ensemble de jobs qui prend un ou plusieurs inputs, et écrit dans une ou plusieurs destinations, avec d'éventuelles transformations.
      - Les pipelines seront souvent liées : par exemple une pipeline d'ingestion qui écrit dans le data lake, puis une autre qui lit cette donnée, la combine avec une autre, et écrit à nouveau dans le data lake.
      - Sa structure est :
        
        ID
        
        Name
        
        Description
        
        Type : indique par exemple si c'est une pipeline d'ingestion ou de transformation.
        
        Velocity : batch ou real-time.
        
        Sources and Destinations : liste les identifiants d'objets Source desquels la pipeline lit, et Destination vers lesquels la pipeline écrit.
        
        En général une pipeline d'ingestion aura une source et une destination, et une pipeline de transformation aura plusieurs sources et une destination.
        
        Data Quality Checks IDs : une liste d'identifiants de checks de qualité à appliquer à l'ensemble des sources et destinations de la pipeline.
        
        Created At
        
        Updated At
        
        Connectivity Details : pour les pipelines d'ingestion, il s'agit d'avoir des informations sur les sources. Par exemple des URLs, adresses IP etc.
        
        Attention à ne pas stocker de username / mots de passe dans ce layer. Il vaut mieux les mettre dans des outils sécurisés comme Azure Key Vault, AWS Secrets Manager ou Google Cloud Secrets Manager.
    - L'objet Source décrit un endroit dont on veut aller chercher de la donnée en entrée d'une pipeline.
      - Sa structure est :
        
        ID
        
        Name
        
        Schema ID : un lien vers le schema registry qui contient le schéma de cette source.
        
        Data Quality Checks IDs : les checks de qualité à appliquer à chaque fois que cette source est utilisée.
        
        Type : le type de source, par exemple “file”, “real-time topic”, “table”.
        
        Created At
        
        Updated At
    - L'objet Destination est similaire à l'objet Source, mais les types peuvent être différents. Par exemple, on peut vouloir aussi mettre dans un key/value store.
      - Sa structure est :
        
        ID
        
        Name
        
        Schema ID
        
        Data Quality Checks IDs
        
        Type
        
        Created At
        
        Updated At
  - 2 - Les Data Quality Checks permettent d'identifier les données qui posent problème.
    - Ils s'appliquent à des pipelines et sources ou destinations sans êtres spécifiques à un namespace.
    - Il existe deux types de data quality checks :
      - Les proactive checks sont faits pour contrôler la donnée une par une, et s'assurer que la donnée de mauvaise qualité ne rentre pas.
        
        On va souvent vérifier le format de la donnée, ou le fait que certaines valeurs soient cohérentes. Par exemple 24h dans un jour, pas de dates négatives etc.
        
        Ces checks ne peuvent pas être trop lourds pour ne pas bloquer la pipeline trop longtemps.
      - Les retrospective checks sont schédulés régulièrement, et opèrent sur de plus grandes quantités de données, pour s'assurer qu'on garde une certaine consistance sur l'ensemble.
        
        Ca peut par exemple être de faire une jointure sur deux jeux de données de départements et d'employés, pour s'assurer qu'aucun département n'est sans employé.
        
        Ils produisent des rapports réguliers pour donner lieu à d'éventuelles actions pour améliorer la qualité de la donnée.
    - L'élément principal du data quality check est la règle à faire respecter. Il existe de nombreuses options sur la manière de l'implémenter.
      - Ça peut être une requête SQL, ou encore un Domain Specific Language (DSL).
    - Leur structure est :
      - ID
      - Name
      - Severity : la gravité du problème si la règle n'est pas respectée.
        
        info indique qu'on laisse passer la donnée, qu'on log le problème dans l'activity metadata, mais qu'on ne crée pas d'alerte.
        
        warning indique qu'on laisse passer la donnée, et qu'on crée une alerte pour avertir un data engineer.
        
        critical indique qu'on ne laisse pas passer la donnée et qu'on la met en quarantaine, avec aussi une alerte.
      - Rule : en fonction de la manière dont on gère nos règles, cet attribut contiendra quelque chose de différent.
      - Created At
      - Updated At
  - 3 - Les Pipeline Activities contiennent des informations de succès, échecs, statistiques etc. sur l'exécution régulière des pipelines.
    - On enregistre les informations de chaque pipeline qui tourne, et on ne supprime jamais ces données, pour pouvoir ensuite investiguer, ou faire des analyses dessus.
    - On pourra par exemple répondre à des questions comme :
      - Quelle est la durée moyenne d'une pipeline ?
      - Combien de rows lit en moyenne une pipeline ?
      - Combien de données on collecte en moyenne depuis une source donnée ?
    - Parmi les éléments de structure que les auteurs ont trouvé utiles dans la plupart des contextes :
      - Activity ID
      - Pipeline ID
      - Start time, Stop time : début et fin de l'exécution de la pipeline.
      - Status : succès / échec.
      - Error Message : en cas d'échec, mettre l'erreur dans ce champ fait gagner beaucoup de temps de recherche dans les logs.
      - Source and Destination Ids : la liste précise des sources et destinations qui ont été utilisées par la pipeline.
      - Rows Read : nombre de rows qui ont été lues, dans le cas de fichiers ça permet notamment de s'assurer qu'on a lu le fichier entier.
      - Rows Written
      - Bytes Read
      - Bytes Written : on peut l'utiliser pour du monitoring, par exemple pour s'assurer que la valeur ne vaut pas 0 si Bytes Read ne vaut pas 0.
      - Extra : des infos additionnelles comme le path où le fichier a été écrit sur le storage, le nom du topic et le window dans le cas de real-time.
    - Dans le cas de real-time processing, c'est une bonne idée d'aligner le time window avec la fréquence d'écriture des messages dans le slow storage.
  - 4 - Schema Registry contient l'ensemble des versions des schémas des données entrantes. Il est détaillé au chapitre suivant.
Selon l'expérience des auteurs, il n'y a pas d'outil open source ou commercial qui permette de mettre en œuvre le metadata layer de manière satisfaisante. Ils conseillent donc de le coder soi-même.
- 1 - Une première solution simple est d'implémenter le metadata layer avec des fichiers.
  - Il s'agit de la solution la plus simple, quand on a peu de sources et de pipelines.
  - La pipeline metadata peut être implémentée avec des fichiers de configuration de type JSON ou YAML par exemple.
    - Il s'agit d'avoir par exemple un fichier pour les namespaces, un pour les pipelines etc.
    - Les IDs doivent être assignés à la main.
    - Il s'agira de les mettre dans le gestionnaire de version avec le reste du code, et de les déployer à chaque fois avec la pipeline de CI/CD.
  - Les pipeline activities metadata sont l'équivalent de fichiers logs où la donnée afflue régulièrement.
    - Pour pouvoir chercher dedans, il faut un outil spécialisé qui permette de le faire, il s'agit des Cloud Log Aggregation Services : Azure Monitor avec Log Analytics sur Azure, Cloud Logging sur GCP, et Elasticsearch sur AWS.
- 2 - Un cran de complexité au-dessus, on a l'utilisation d'une base de données pour stocker les fichiers de configuration (la pipeline metadata).
  - Les fichiers de configuration sont toujours dans le gestionnaire de version, et servent de source de vérité pour la configuration du metadata layer. C'est nécessaire pour avoir un historique des changements.
  - A chaque fois qu'un changement est fait dans ces fichiers, une migration sera faite sur la metadata database.
  - L'avantage d'avoir cette DB, c'est qu'on va pouvoir faire des requêtes pour obtenir des informations spécifiques qui existent à travers les fichiers de config. Par exemple : “Je veux voir toutes les sources qui utilisent ce data quality check”.
  - La DB peut être une DB relationnelle ou une DB de document qui permettra plus de flexibilité sur l'évolution du schéma.
    - Des exemples typiques peuvent être Google Cloud Datastore, Azure Cosmos DB et AWS DynamoDB.
- 3 - Quand on a plusieurs équipes en charge des pipelines, il faut une solution qui puisse abstraire les détails d'implémentation exposés par la DB : on peut utiliser une metadata API.
  - L'idée c'est que le changement dans la structure de la DB n'impactera pas de nombreux outils maintenus par plusieurs équipes différentes. On pourra par exemple faire plusieurs versions de l'API.
  - La metadata API est en général faite selon les principes REST.
    - Pour plus d'infos sur comment designer une API REST, il y a The Design of Web APIs d'Arnaud Lauret.
  - Il faudra que l'ensemble des outils qui utilisaient la DB, y compris les pipelines, utilisent maintenant l'API pour accéder aux configurations.
- Les auteurs conseillent de commencer par implémenter la solution la plus simple qui satisfait les besoins actuels de la data platform, avec la possibilité de passer à la version un cran plus complexe dès que le besoin sera là.
  - Chaque solution se base sur la précédente en lui ajoutant quelque chose, donc ça devrait être relativement facile de migrer.
Parmi les outils qu'on peut trouver chez les cloud vendors, qui se rapprochent le plus de ce qu'on recherche avec notre metadata layer, il y a :
- AWS Glue Data Catalog stocke des informations à propos des sources et destinations, et des statistiques sur les runs des pipelines, ce qui fait de cet outil le plus proche de ce qu'on recherche.
  - Le désavantage majeur c'est sa flexibilité : il faut implémenter les pipelines avec AWS Glue ETL, ce qui veut dire n'avoir que des batch jobs, et qui soient compatibles avec Glue (donc pas de source REST par exemple).
- Azure Data Catalog et Google Cloud Data Catalog sont plus orientées business metadata, et fournissent surtout de la data discovery : permettre aux utilisateurs de la donnée de faire une recherche dans une UI pour trouver la table qui les intéresse.
Parmi les outils open source, qui se rapprochent le plus de ce qu'on recherche avec notre metadata layer, il y a :
- Apache Atlas permet de faire de la data discovery, mais aussi de gérer la configuration de pipelines de manière flexible : on peut utiliser les Types qu'il propose pour créer la configuration des namespaces, des pipelines, sources, destinations etc. avec des liens entre les objets.
  - Son inconvénient principal est qu'il a été créé pour l'écosystème de Hadoop, et possède de nombreuses fonctionnalités qui lui sont dédiées.
  - Un autre inconvénient est que c'est un outil open source qui nécessite de faire tourner d'autres outils open sources difficiles à administrer : HBase et Solr.
- DataHub est similaire à Atlas, dans la mesure où il est suffisamment flexible pour permettre d'implémenter le modèle décrit dans ce chapitre, et permet aussi la data discovery.
  - Il a aussi l'inconvénient de nécessiter de faire tourner des outils open source difficiles à administrer : Kafka, MySQL, Elasticsearch et Neo4j.
- Marquez permet principalement de mettre à disposition des informations de data lineage, c'est-à-dire des informations sur l'origine des données.
  - Il n'est pas assez flexible pour implémenter le modèle présenté dans ce chapitre.
  - Il a l'avantage de ne nécessiter que PostgreSQL comme dépendance à faire tourner, et on peut le faire comme service managé.

8 - Schema management

Certaines organisations ont une approche proactive, et planifient les conséquences des changements dans les DBs opérationnelles sur les équipes data.
- D'autres ont une approche ”do nothing and wait for things to break”, et attendent simplement que la pipeline ETL casse pour que l'équipe data la répare en prenant en compte le changement de schéma.
Dans les architectures data traditionnelles basées sur le data warehouse, les données arrivent dans une landing table qui reproduit exactement leur schéma, et donc quand elles changent, l'ingestion casse.
- Il existe une approche alternative schema-on-read où il s'agit d'ingérer la donnée telle quelle dans un système de fichiers distribués, et dans ce cas on repousse le problème au processing layer.
Coupler le schema-on-read avec une approche ”do nothing and wait for things to break” est plutôt une mauvaise idée selon les auteurs. Comme alternatives, on a :
- 1 - Le schema as a contract où il s'agit pour l'équipe de développeurs d'enregistrer le schéma de leur source de donnée dans le schema repository, et d'en être responsables.
  - Ils doivent alors ne faire que des changements backward-compatibles dans leur DB. Par exemple ajouter des colonnes mais pas en renommer.
  - Pour que ça marche, il faut deux choses :
    - Un grand niveau de maturité dans les process de développement, notamment d'un point de vue automatisation de du check de rétrocompatibilité dans la pipeline de CI.
    - Un owner pour chaque source de donnée externe à l'organisation.
  - De l'expérience des auteurs, les organisations n'ont en général pas la maturité technique suffisante, et le besoin de schéma versionné venant après coup, il est difficile de convaincre les équipes opérationnelles de mettre en place le schema as a contract.
  - NDLR : il s'agit de l'approche mise en avant par le Data Mesh.
- 2 - La gestion du schéma dans la data platform. Dans ce cas, la responsabilité se trouve du côté de l'équipe qui gère la data platform.
  - Les auteurs trouvent que cette solution marche bien dans pas mal de contextes. Elle a l'avantage de permettre de centraliser au même endroit les schémas des données qui viennent des équipes internes et ceux qui viennent de l'extérieur.
    - Cette centralisation permet ensuite d'avoir un catalogue de données dans lequel on peut fouiller.
    - Ca permet d'avoir un historique des schémas pour pouvoir utiliser n'importe quelle donnée archivée, ou faire du debugging.
    - Ca peut aussi permettre de détecter et ajuster les changements de schémas avant que la pipeline n'échoue.
  - Une autre solution peut être de laisser aux équipes internes la responsabilité du schéma de leurs données, et de centraliser les schémas des sources externes chez l'équipe responsable de la data platform.
Dans le cas où la gestion des schémas se fait dans la data platform, elle doit être ajoutée en tant que 1ère étape du common data processing.
- Le module de schema-management va d'abord vérifier si un schéma existe déjà pour cette source.
  - S'il n'existe pas, le module va inférer un schéma depuis les données, puis enregistrer ce schéma dans le schéma registry en tant que V1.
  - S'il existe, le module va récupérer la dernière version depuis le schema registry, puis inférer le schéma depuis les données, créer un nouveau schéma compatible avec les deux et l'enregistrer en tant que version actuelle.
- L'inférence de schéma dont on est en train de parler se base sur Apache Spark.
  - Spark est capable d'inférer le schéma de fichiers CSV, JSON, y compris s'il y a plusieurs records dedans.
  - Il utilise un sample de records pour faire l'inférence, par défaut 1000, et ce nombre est configurable.
    - S'il est trop faible on risque d'avoir une inférence faussée qui ne permet pas de parser l'ensemble des données. Et s'il est trop grand on risque d'avoir des problèmes de performance.
    - Pour une table d'une DB relationnelle par exemple, le nombre pourra être bas parce que la schéma est garanti par la DB.
  - Dans le cas où la donnée est différente entre deux records, Spark essayera de trouver un type qui englobe les deux. Par exemple, un nombre et un string vont donner un string.
  - Dans le cas où un type commun n'est pas possible, les données minoritaires seront placées dans le champ _corrupt_record.
  - Le schéma inféré par Spark va d'abord être converti en schéma Avro avant d'être mis dans le schema registry.
  - Si on utilise un outil qui ne supporte pas l'inférence de schéma, comme par exemple Google Cloud Dataflow basé sur Apache Beam, alors il faudra gérer les schémas à la main.
- Dans le cas d'une real-time pipeline, on ne peut pas utiliser l'inférence à cause du problème de performance et de la quantité de schémas qui seraient générés.
  - Dans ce cas, les auteurs conseillent de laisser les développeurs qui génèrent les données de streaming maintenir le schéma.
- Pour pouvoir avoir du monitoring sur les changements de schémas, le module de schema-management peut créer un log dans la partie Pipeline Activities du metadata layer à chaque fois qu'il trouve des données avec un schéma qui a changé.
  - Même si l'ingestion et les common data processing steps peuvent se “réparer” automatiquement, la suite du processing peut ne pas donner le résultat voulu. Par exemple un rapport qui n'a plus les valeurs d'une colonne qui a été enlevée par la source.
  - Il vaut mieux être alerté du changement de schéma, et prévenir les équipes qui utilisent les données de cette source, avant qu'ils ne s'aperçoivent du problème par eux-mêmes.
Côté implémentation du schema registry :
- Apache Avro est l'option conseillée par les auteurs pour servir de format de base pour l'ensemble des données de la data platform.
  - Son schéma peut être écrit et maintenu à la main.
  - Spark peut aussi transformer son schéma inféré en schéma Avro automatiquement.
  - Ces schémas peuvent être représentés par du simple JSON, et donc n'importe quelle DB qui supporte ça peut les héberger.
  - Avro a un très bon système de gestion des versions des schémas.
- Les solutions cloud-natives de type data catalog permettent d'implémenter un schema registry, mais ont des limitations.
  - La plupart sont surtout orientés data discovery, et manquent de fonctionnalités concernant la gestion des versions des schémas et le support d'Avro.
  - Confluent Schema Registry offre les fonctionnalités de gestion de version des schémas et un bon support d'Avro, mais il nécessite d'utiliser Kafka, ou un outil compatible avec Kafka.
    - Donc si on utilise par exemple Kinesis, ou bien si on ne fait pas de real-time, on ne pourra pas utiliser leur schema registry.
- La solution maison proposée par les auteurs consiste à avoir soit une DB, soit une API avec une DB derrière.
  - Le solution pure texte stockée dans le gestionnaire de version, similaire au reste de la configuration du metadata layer, ne peut pas marcher pour le schema registry parce qu'il faut pouvoir le mettre à jour automatiquement.
  - Comme DB, on peut utiliser les mêmes Cosmos DB, Datastore et DynamoDB.
  - La structure des objets de schéma sera :
    - ID
    - Version : l'ID et la Version forment ensemble une clé unique. L'ID en elle-même n'est donc pas unique pour éviter d'avoir à mettre à jour en permanence les configurations des sources et destinations.
    - Schema : le champ qui stocke le schéma Avro au format texte.
    - Created At
    - Updated At
Concernant la stratégie de gestion de version des schémas.
- Il existe deux types de compatibilité entre les schémas :
  - 1 - Backward-compatible : la dernière version du schéma doit permettre de lire l'ensemble des données existantes, y compris produites par un ancien schéma.
    - Avro impose des règles précises pour garder la backward-compatibility. Par exemple ajouter une colonne le permet, dans ce cas la lecture d'une donnée ancienne par un schéma récent donnera lieu à l'usage de la valeur par défaut pour la colonne manquante.
  - 2 - Forward-compatible : une version plus ancienne du schéma doit permettre de lire les données produites par une version plus récente.
    - Si on reprend l'exemple de l'ajout de colonne, Avro permet une forward-compatibility : l'ancien schéma ignorera la nouvelle colonne au moment de la lecture de la nouvelle donnée.
  - Le renommage de colonne est l'équivalent d'une création de colonne, et d'une suppression de colonne. Donc si on a des valeurs par défaut dans le schéma, elle sera à la fois backward-compatible et forward-compatible.
  - Avro supporte aussi automatiquement certains changements de types, par exemple un entier 32 bits vers 64 bits. On peut aussi soi-même implémenter d'autres règles de conversion, mais les auteurs le déconseillent pour garder la complexité des pipelines faible.
  - Les règles d'évolution de schéma d'Avro sont disponibles dans leur doc (opens in a new tab).
- Les common data transformation pipelines ne vont en général pas avoir besoin de la présence de colonnes spécifiques, et donc vont être résilientes aux changements de schémas.
  - Les business processing pipelines en revanche vont y être beaucoup plus sensibles.
- Les auteurs conseillent d'utiliser les anciens schémas dans les pipelines, et de passer aux nouveaux quand les changements de code ont été faits. Ca veut dire s'efforcer à faire des changements de schémas forward-compatibles.
  - Quelle que soit la stratégie, même si la pipeline ne casse pas grâce aux règles de backward / forward compatibility, il est possible qu'on se retrouve avec des erreurs logiques dans nos transformations.
    - Par exemple, une colonne indiquant le nombre de ventes est renommée, et peut continuer à être lue de manière forward compatible avec la valeur par défaut NULL. Mais le dashboard se mettra à montrer une absence de ventes.
    - Il n'y a pas de solution simple à ce problème. Il faut avoir un système de monitoring et d'alerting efficaces, et prévenir les clients en amont que leurs dashboards risquent d'avoir des incohérences le temps de mettre à jour le code.
Alors que les fichiers peuvent avoir chacun leur version de schéma associée, la donnée qui se trouve dans une table du data warehouse ne peut pas avoir plusieurs schémas en même temps.
- On ne peut pas simplement utiliser le schema registry pour mettre à jour la table du data warehouse, il va falloir le faire avec du code dans le module de schema-management, qui fait partie des common data transformations.
  - Ça veut dire que les changements de schéma se feront quand même de manière automatique, avec des règles pré-établies où on va générer le bon SQL pour restructurer la table, en fonction de chaque changement de schéma Avro.
- On ne peut pas non plus appliquer les mêmes règles qu'avec les transformations de schémas entre fichiers : dans le cas de suppression d'une colonne (ou de renommage, qui implique une suppression de fait), on va garder l'ancienne colonne quand même pour garder la donnée historique.
  - Parfois, quand les données ne sont pas trop grosses, il pourra être préférable de supprimer la colonne et de la recréer avec les données historiques et les nouvelles données dedans.
- Côté data warehouses des cloud vendors :
  - AWS Redshift et Azure Synapse ont une approche similaire :
    - Ils sont ancrés dans le monde du relationnel, et nécessitent la définition du schéma avant de charger de la donnée.
    - Ils supportent ALTER TABLE pour faire des changements sur les tables.
    - Redshift supporte Avro mais sans inférence à partir du schéma, alors que Synapse supporte seulement CSV, ORC et Parquet.
  - Google BigQuery a une approche moins relationnelle, et permet d'inférer le schéma à partir de la donnée qu'on lui donne.
    - Il va aussi ajouter des colonnes au schéma automatiquement en inférant le type, si on lui présente de la donnée qui a des colonnes en plus. Il le supporte pour Avro, JSON et Parquet.
    - En revanche, il ne permet pas de modifier les tables après coup, sauf en ajoutant ou supprimant des colonnes, ce qui peut prendre du temps et coûter cher.

9 - Data access and security

Les données d'analytics sont utilisées par de plus en plus de personnes au sein des entreprises, et par des moyens variés.
- 1 - Il y a les utilisateurs humains qui utilisent en général des outils BI ou veulent exécuter des requêtes SQL, et parfois des data scientists qui veulent accéder à la raw data.
- 2 - Et il y a les applications qui utilisent la donnée par exemple pour des applications ML de recommandation ou de prédiction. Le data warehouse ne suffit pas pour ces cas d'usage.
Le data warehouse reste quand même l'outil le plus commun pour accéder à la donnée d'analytics, du fait de la compatibilité avec les outils BI et du support du SQL.
- AWS Redshift.
  - Il s'agit d'un data warehouse distributé, c'est-à-dire qu'il répartit la donnée sur plusieurs machines.
    - Un nœud leader reçoit les requêtes et répartit le travail à faire et les données sur les autres nœuds.
    - Les autres nœuds eux-mêmes sont subdivisés en slices. Ces slices peuvent être déplacés de nœud en nœud, pour équilibrer la capacité par du rebalancing.
    - Quand on crée une table, on peut indiquer sa propriété DISTSTYLE pour choisir la manière dont ses données seront distribuées sur les nœuds. C'est le réglage de performance le plus impactant.
      - ALL : une copie de la table est créée sur chaque nœud. On ne peut le faire qu'avec les petites tables qui sont souvent l'objet de jointures.
      - EVEN : les rows de la table sont répartis de manière équitable sur les nœuds.
      - KEY : permet d'indiquer une colonne dont les valeurs identiques donneront lieu à ce que la donnée soit stockée sur la même machine.
      - AUTO : vaut ALL au début, et passe à EVEN quand la table grandit.
  - Il est basé sur PostgreSQL et présente les caractéristiques des DB relationnelles.
    - Il ne supporte que les types “primitifs”, c'est-à-dire pas les tableaux ou les objets imbriqués. Il est donc peu adapté à de la donnée JSON, avec laquelle les optimisations d'encodage ou de distribution dans les nœuds par clé ne pourront pas servir.
  - On peut optimiser la taille des données en choisissant le type d'encodage pour les colonnes : par exemple dans le cas où une colonne peut avoir seulement quelques valeurs possibles, l'encodage byte-dictionary permet de limiter la taille de ces données.
  - Il possède une fonctionnalité appelée Spectrum, qui permet de créer une table dans Redshift, dont les données sont sur S3.
    - Ca permet d'éviter d'utiliser des ressources CPU et de l'espace sur le data warehouse, pour des données qu'on veut juste explorer par exemple.
    - Les performances seront du coup moins bonnes que les données qui sont sur les nœuds Redshift.
    - Les auteurs recommandent de créer une DB dédiée sur Redshift pour regrouper ces tables qui pointent vers ailleurs.
- Azure Synapse.
  - C'est une DB distribuée comme Redshift, avec un control node principal qui reçoit les requêtes, et qui fait appel aux autres nœuds.
    - Il y a une séparation storage / compute. Les données sont séparées en 60 distributions, et sont associées à des compute nodes.
    - Il n'est pas complètement élastique, puisque pour redimensionner le cluster, il faut tout arrêter, et ça peut prendre du temps.
    - Les tables peuvent être configurées pour la répartition de leurs données, de la même manière que Redshift.
      - REPLICATE : l'équivalent de ALL, c'est-à-dire copier sur chaque nœud.
      - ROUND ROBIN : l'équivalent de EVEN, c'est-à-dire répartir entre les nœuds.
      - HASH : l'équivalent de KEY, c'est-à-dire spécifier une colonne dont les valeurs permettront de répartir les données.
  - Il présente des caractéristiques relationnelles.
    - Il supporte seulement les types primitifs, et fournit des fonctions de parsing pour JSON, mais au prix de nombreuses optimisations perdues.
  - Il a une fonctionnalité similaire à Spectrum, configurable par la notion de pools.
    - SQL pool représente l'utilisation normale du data warehouse.
    - SQL on-demand pool permet de faire des requêtes sur des données sur Azur Blob Storage au format Parquet, CSV ou JSON.
    - Spark pool permet de faire des requêtes avec Spark, sur des données qui sont dans Azur Blob Storage. Ils permettent l'auto-scaling, mais nécessitent que 3 nœuds tournent en permanence.
- Google BigQuery.
  - BigQuery est un peu plus “managé” que les deux autres, dans la mesure où il n'y a pas de besoin de planifier la capacité dont on aura besoin à l'avance.
    - La puissance de calcul est “provisionnée” à chaque requête, grâce à des groupes de dizaines de milliers de nœuds qui tournent en permanence dans l'infra de Google.
    - Comme il est plus managé, on peut aussi moins facilement contrôler la manière dont les données d'une table sont réparties au sein des nœuds.
      - On a quand même la notion de partitioning qui permet de répartir les données selon les valeurs d'une colonne.
      - Et de clustering qui permet d'organiser physiquement les données de manière à rendre les requêtes qu'on fait le plus souvent plus efficaces.
    - Le pricing se fait aussi sur la quantité de données traitée, ce qui peut être avantageux quand on a de petits besoins, mais rend les coûts difficilement prédictibles.
  - Les nœuds de calcul sont sur des machines différentes des nœuds de stockage : on n'a pas de data locality.
    - C'est moins rapide que si la donnée était locale, mais ça évite d'avoir à recopier la donnée à chaque rebalancing. La donnée est accédée via le réseau local de Google qui est suffisamment performant pour que ça passe.
  - BigQuery vient initialement plutôt d'un système permettant de traiter des fichiers de log, et non pas un système relationnel comme les deux autres.
    - Il a un support natif des structures imbriquées, et peut traiter le JSON comme une structure et pas juste du texte, avec la possibilité d'appliquer des traitements sur les attributs.
    - Il est du coup moins facilement compatible avec les outils BI, il faudra passer par une API REST.
- Les grandes organisations peuvent tirer parti de l'utilisation de plusieurs cloud providers, mais pour les petites le coût opérationnel n'en vaut pas la peine. Le choix du data warehouse dépendra donc en général du choix du cloud provider pour le reste de l'infra.
Les applications utilisent de plus en plus la data dans des systèmes customer-facing, par exemple dans des systèmes de recommandation.
- Donner à l'application un accès au data warehouse serait une mauvaise idée pour plusieurs raisons :
  - Les data warehouses ne sont pas conçues pour offrir des latences se comptant en millisecondes, mais en secondes voire minutes sur de grandes quantités de données.
  - Ils ne sont pas conçus pour supporter un trop grand nombre de transactions en même temps (par exemple des dizaines ou centaines de milliers) comme pourrait le nécessiter une application.
  - Si l'application est compromise, l'ensemble du contenu du data warehouse pourrait fuiter, alors que si l'application a seulement accès à une DB qui a ce dont elle a besoin, on aura une meilleure sécurité.
- 1 - Cloud relational databases.
  - Chaque cloud provider a ses services de DBs managées, qui tiennent sans problèmes jusqu'à 1 TB. Au-delà de ça, ou si on a besoin de situer les machines géographiquement, il faut une DB distribuée.
  - AWS propose Relational Database Service (RDS) pour PostgreSQL, MySQL, MariaDB, Oracle et SQL Server.
    - Il propose Aurora comme DB distribuée, compatible avec MySQL et PostgreSQL.
  - GCP propose Google Cloud SQL, qui supporte MySQL, PostgreSQL et SQL Server.
    - Il propose Google Cloud Spanner pour la version distribuée.
  - Azure propose Azure SQL Database, qui supporte MySQL, PostgreSQL et SQL Server.
    - Il propose HyperScale pour la version distribuée, disponible seulement pour SQL Server.
- 2 - Cloud key / value data stores.
  - Les services key/value offrent une faible latence pour insérer et retrouver des valeurs par leur clé.
    - Ils sont souvent utilisés par les nouveaux projets pour pouvoir itérer vite sans avoir de migration à faire.
  - Les cloud providers proposent soit une version pay per use plus avantageuse en cas de faible utilisation, et une version pay per provisioned capacity plus avantageuse en cas de grosse utilisation.
  - AWS propose DynamoDB, qui reste performant quel que soit le scale, et offre les deux types de facturation.
  - GCP propose Datastore qui est similaire à DynamoDB et qui propose du pay per use, et Cloud Bigtable qui ne permet pas de mettre de contrainte de types sur les données, et supporte le price per provisioned capacity.
    - Cloud Bigtable est d'ailleurs compatible avec HBase.
  - Azure propose CosmosDB, qui a la particularité de supporter les API clientes de MongoDB, Cassandra, SQL et de graph API, ce qui rend le portage depuis ces technos facile.
- 3 - Full-text search services.
  - Dans le cas où la fonctionnalité de notre application est de permettre une recherche dans la donnée, il existe Solr et Elasticsearch, tous deux basés sur Lucene.
    - Par exemple, si on veut chercher quelque chose de similaire à ce qui est tapé par l'utilisateur.
  - AWS propose CloudSearch, Azure propose Azure Search, et GCP ne propose rien de managé au moment de l'écriture du livre.
- 4 - In-memory cache.
  - Les caches permettent des temps d'accès inférieurs à la milliseconde grâce au stockage en RAM. Ils doivent être liés à une DB persistante pour pouvoir être reconstruits.
  - AWS propose ElasticCache, qui supporte Memcached et Redis, GCP propose Memorystore qui supporte Memcached, et Azure propose Azure Cache qui supporte Redis.
Les modèles de machine learning nécessitent l'accès à une grande quantité de données variée, une grande puissance de calcul, et l'accès à des outils spécifiques. La cloud data platform est parfaitement adaptée à ça.
- Dans les plateformes traditionnelles, les data scientists passent 80% de leur temps à récupérer la donnée sur leur machine, et la nettoyer et la transformer pour qu'elle puisse être interprétée par leurs outils.
  - Ils vont ensuite faire des tests exploratoires pour comprendre ce qu'ils peuvent faire ce cette donnée.
  - Puis ils séparent la donnée en deux : la donnée d'entraînement et la donnée de validation.
  - Ils vont faire un cycle entraînement / validation où ils vont plusieurs fois améliorer le modèle puis le tester contre la donnée de validation.
    - Cette validation permet d'éviter l'overfitting où le modèle ne serait bon que sur les données avec lesquelles il s'est entraîné.
    - Faire l'entraînement sur leur machine locale leur prend beaucoup de temps.
  - Une fois que le modèle est fonctionnel, il faut le rendre production-ready pour le déployer, en ajoutant du logging, de la gestion d'erreurs etc. ce qui est souvent difficile.
- La cloud data platform aide au développement de modèles ML.
  - Une bonne partie de la mise en forme et de la validation des données est faite dans l'ingestion layer et dans le processing layer avec les common data transformation steps.
  - Les data scientists peuvent copier la donnée comme ils veulent dans le storage de la cloud data platform, et faire de l'exploration ou du processing sans télécharger les données en local.
  - Ils peuvent collaborer sur un même jeu de données puisqu'il est dans le cloud, et peuvent avoir accès à de la donnée de production en grande quantité.
  - Chacun des cloud vendors fournit un service de ML permettant de gérer un projet ML de bout en bout, et de mieux collaborer entre data scientists : SageMaker chez AWS, AI Platform chez GCP, et Azure ML chez Azure.
La business intelligence et le reporting sont en général le premier usage de la donnée de type analytics.
- Ces outils nécessitent souvent que la donnée soit relationnelle, c'est-à-dire que chaque donnée soit dans sa colonne avec la table “à plat” reliée à d'autres tables par des clés étrangères, plutôt que d'avoir des données imbriquées comme dans du JSON.
  - BigQuery commence à être supporté par des outils comme Tableau, mais tous ne le supportent pas correctement.
- Bien que de nombreux outils BI supportent Spark SQL, et pourraient se brancher directement sur le data lake, les auteurs le déconseillent parce que ça rendrait l'interface de ces outils peu interactive et lente. Se brancher sur le data warehouse est bien plus adapté pour cette raison.
- Excel peut se brancher sur le data warehouse grâce à son API JDBC/ODBC, mais c'est un outil qui tourne sur une machine locale, donc il sera limité sur la quantité de données, et télécharger les données sur sa machine locale pose des problèmes de performance.
- On voit souvent des outils externes, par exemple chez d'autres cloud providers, accéder à la donnée de la cloud data platform.
  - Il faut faire attention aux coûts de sortie des données (data egress costs), que chaque cloud provider applique.
  - Chaque cloud provider a sa solution BI : Azure Power BI qui est très connu, AWS QuickSight, et DataStudio et Looker BI pour GCP.
La sécurité est essentielle pour une plateforme data.
- Il vaut mieux éviter les accès ad hoc dès qu'il y a un besoin, mais plutôt utiliser les concepts de Users, Groups et Roles fournis par les cloud providers.
  - Les groupes facilitent grandement la gestion des permissions, il vaut mieux les configurer à ce niveau là dans la mesure du possible.
  - Une bonne pratique est de ne fournir que les permissions nécessaires à chaque type d'utilisateur (principle of least privilege).
- Il existe des outils cloud-native pour l'authentification, à la place des mots de passe, par exemple Azure Active Directory. Les auteurs conseillent de les utiliser quand c'est possible.
- Certaines configurations permettent de rendre des services accessibles publiquement. Pour limiter le risque, on peut faire diverses choses comme des audits, ou l'utilisation du principe infrastructure-as-code.
- Dans le cas où on a des données sensibles, il ne faut pas hésiter à chiffrer des colonnes particulières.
- Une autre solution peut être de limiter l'accès réseau à la donnée, dans le cas où les utilisateurs seraient sur un réseau particulier.

10 - Fueling business value with data platforms

La data platform doit être organisée autour d'une data strategy, c'est-à-dire être au service des objectifs business.
- Parmi les grands objectifs business, on trouve :
  - Augmenter les revenus.
  - Augmenter l'efficacité opérationnelle.
  - Améliorer l'expérience utilisateur.
  - Permettre l'innovation.
  - Améliorer la conformité.
- Exemples :
  - Dans le cas d'une entreprise de jeux vidéo qui veut maximiser les achats ou la publicité in-game, la stratégie peut être d'optimiser la plateforme data pour du real-time processing des événements du jeu.
  - Dans le cas d'une entreprise minière qui veut réduire ses coûts opérationnels, la stratégie peut être d'optimiser la plateforme pour ingérer la donnée des capteurs des engins miniers, et prédire quand faire la maintenance.
La maturité data d'une organisation passe par 4 étapes.
- 1 - See : le business veut voir des rapports et des dashboards pour mieux comprendre ce qui se passe par rapport à ce qui s'est passé dans le passé.
  - Traditionnellement les rapports sont créés par des personnes spécialistes de ces outils, à la demande du business.
  - Dans les plateformes modernes, on applique le principe Bring Your Own Analytics (BYOA), où les personnes du business utilisent leurs propres outils qu'ils branchent sur la data platform, pour créer leurs rapports.
  - Ces outils sont branchés sur le data warehouse.
- 2 - Predict : une fois qu'on a ce qui s'est passé et se passe, on veut prédire ce qui va se passer, par exemple avec du ML.
  - Il faut que la plateforme puisse proposer une grande quantité de données.
  - Les données brutes vont être plutôt sur le data lake, et les données raffinées sur le data warehouse.
- 3 - Do : on va donner le résultat des deux premières étapes à des systèmes pour déclencher des actions.
  - Ça peut être du ML avec par exemple des systèmes de recommandation, ou même simplement de la donnée qui est déplacée vers le système opérationnel pour servir les clients.
  - Le fait de déplacer des données du monde analytics au monde opérationnel s'appelle l'orchestration.
  - Ça implique que le système qui utilise cette donnée soit disponible et réponde aux exigences d'un système de production.
- 4 - Create : la donnée initialement collectée comme analytics devient la source pour un nouveau produit.
  - Par exemple, une banque qui a collecté des données pour améliorer l'expérience utilisateur en aidant les agents à anticiper les réactions des clients, s'est rendue compte qu'elle pouvait l'utiliser aussi pour améliorer l'app mobile.
  - Autre exemple, une entreprise de sécurité s'est servie des dashboards construits pour visualiser les intrusions, pour montrer aux clients en quoi elle leur apportait de la valeur avec tous les risques qu'elle a évités.
Certains challenges non techniques peuvent faire échouer la cloud data platform.
- Délivrer de valeur business rapidement : le business a besoin d'itérations qui résolvent de vrais besoins au bout de quelques mois maximum.
  - Les auteurs conseillent de partir d'un use-case pas trop complexe lié à la data, et de l'implémenter en faisant avancer la plateforme. Et on passe comme ça de use-cases en use-cases.
  - L'alternative moins intéressante c'est d'ingérer toutes les sources possibles, pour finir par trouver des cas d'usage avec les sources qu'on supporte déjà.
- Faire adopter la plateforme par les utilisateurs : les utilisateurs ont peut-être déjà leur manière de travailler avec les analytics, en particulier la production de rapport traditionnelle plutôt que la data self-service.
  - Il y a les early adopters qui supportent la nouvelle plateforme, les blockers qui montrent leur scepticisme publiquement, les chickens qui ont peur de tout ce qui est nouveau, et les avoiders qui ne veulent pas toucher à ce qui est nouveau.
  - Quelques conseils pour avoir une meilleure adoption :
    - S'assurer que les premiers utilisateurs soient à la fois des early adopters et des employés influents.
    - Après les early adopters, on peut aider un blocker pour tenter de le retourner. Si ça marche c'est excellent pour le projet.
    - Les chickens ont besoin de beaucoup de formation.
    - Les avoiders mettront plus de temps, mais c'est OK.
    - Ce serait bien d'avoir un sponsor C-level qui soutient le projet, et crée de la visibilité pour lui.
      - Exemple : Disney avait lancé un concours interne où les utilisateurs data pouvaient montrer leurs résultats avec la plateforme et être récompensés.
- Obtenir et garder la confiance des utilisateurs : il faut que la qualité de la donnée soit suffisamment bonne pour que les utilisateurs aient confiance en elle.
  - Prendre en compte l'utilisateur de la donnée et ses besoins rendre dans le cadre de la data governance.
  - Parmi les métriques de qualité à surveiller, il peut y avoir le pourcentage de données correctes, les champs obligatoires remplis, la précision, la consistance, l'intégrité de la donnée etc.
  - Quand la qualité qu'on s'est fixée n'est plus respectée, il faut :
    - 1 - prévenir les consommateurs de la donnée.
    - 2 - Mettre l'équipe sur le coup pour régler le problème au plus vite.
- Éviter la formation de d'un silo autour de la data platform : la responsabilité de la donnée, des règles de qualité et la mesure de la qualité, les SLA etc. doivent être drivées par le business.
  - La donnée part des sources potentiellement extérieures, et irrigue l'organisation à travers divers systèmes. On ne peut pas considérer que la responsabilité de l'équipe technique s'arrête au moment où la donnée sort de la plateforme.
  - Il faut constituer des équipes pluridisciplinaires capables de prendre en charge la responsabilité du système de bout en bout : le fonctionnement de la plateforme et l'utilisation de la donnée.
- Prendre en compte les coûts : pour que la plateforme soit un succès, il faut adopter le point de vue de l'entreprise.
  - Il faut s'intéresser aux manières d'optimiser les coûts des services cloud (FinOps), et comprendre les trade-offs qui y sont liés.
    - Et à l'inverse s'intéresser à ce que la plateforme permet de rapporter d'un point de vue business.
  - Exemple : une grande entreprise de télécommunications recueille des données IoT.
    - La bonne pratique est de faire le processing dans le data lake avec Spark, mais il se trouve que le business avait un deal avec GCP pour une utilisation illimitée de BigQuery à prix fixe.
    - Dans ce cas, la bonne chose à faire sera sans doute de faire des concessions sur le design, et faire le processing dans le data warehouse.

Continuous Discovery Habits Designing Data-Intensive Applications