GESTION DE GRANDES MASSES DE DONNEES - CM1 LE SCHEMA GLOBAL EST CONSTITUE

QUEL OBJECTIF : Schéma conceptuel global :

- Contenant la description globale et unifiée de toutes les données de la
Offrir à l’utilisateur final un accès satisfaisant aux données selon le
BDR
paradigme le plus adapté au besoin.
- Indépendant de la répartition des données
QU’EST-CE QU’UN ACCES SATISFAISANT : Schéma de placement (d’allocation) :
Les requêtes/traitements aboutissent à un résultat avec une latence - Contenant les règles de correspondance avec les données locales
acceptable pour l’usage - Indépendant à la fragmentation et à la réplication
QU’EST-CE QUE LE PARADIGME LE PLUS ADAPTE AU BESOIN : DISTRIBUTION DES DONNEES
Le paradigme qui offrira les meilleures performances, avec le niveau de
MIGRATION :
fiabilité (des résultats ou du système) souhaité par l’utilisateur, pour
effectuer les traitements Transfère d’une relation complète sur un site distant
Intérêt :
LES BASES DE DONNEES PARALLELES - Rapprocher les données du besoin
Principe : SQL :
- Orchestration des calculs d’une requête à travers plusieurs CPUs et - Création d’un wrapper et d’une ‘foreign table’ pour pouvoir faire en
disques avec une gestion de la mémoire adaptée. local une copie de la BD source à la BD cible
Verrous :
REPLICATION/DUPLICATION :
- Gestion de l’équilibrage de charge
- Complexité Création d’une copie conforme d’une table (ou ensemble de tuples) sur un
Différents types d’architectures : site distant. La copie doit rester cohérente avec les données sources
- Architecture à Mémoire Partagée Intérêt :
- Architecture à Disque Partagé - Rapprocher les données des besoins : avoir les mêmes données sur
- Architecture à Mémoire Distribuée différents sites
SQL :
ARCHITECTURE A MEMOIRE PARTAGEE (SHARED MEMORY) - Utilisation d’un processus de Publication / Suscription afin de
Modèle de programmation : synchroniser les tables
- Mémoire partagée (multiprocessus, multithreading) FRAGMENTATION :
Avantages :
- Simplicité, équilibrage de la charge Décomposition d’une relation en plusieurs fragments qui sont transférés
Inconvénients : sur différents sites distants
- Coût du Réseau (bus parallèle ou multibus), faible extensivité Intérêt :
- Découper logiquement les données
ARCHITECTURE A DISQUE PARTAGE (SHARED DISK) SQL :
Avantages : - Création d’un wrapper et d’une ‘foreign table’ pour pouvoir faire en
- Coût du Réseau, extensibilité, migration depuis les monoprocesseurs, local une copie partielle de la BD source à la BD cible en intégrant les
sûreté de fonctionnement contraintes de sélection.
Inconvénients :
- Complexité DIFFERENTES FRAGMENTATIONS

ARCHITECTURE A MEMOIRE DISTRIBUEE (SHARED NOTHING) FRAGMENTATION HORIZONTALE - SELECTION (WHERE)

Avantages :
- Coût, extensibilité, disponibilité
Inconvénients :
- Complexité, équilibrage de charge
BASE DE DONNEES REPARTIE
Définition :
C’est une base de données logique dont les données sont stockées dans
différents SGBD interconnectés
CARACTERISTIQUES
Fragmentation horizontale selon les secteurs
Une base de données répartie dispose d’un schéma global et d’un schéma
PERSONNE_N : SELECT * F ROM PERSONNE WHERE SECTEUR = "N"
d’allocation a distribution des données est transparente pour l’utilisateur
REMARQUE IMPORTANTE
Besoin de formaliser les éléments de la base de données répartie : '
PRINCIPE DE DECOMPOSITION D’UNE BASE DE DONNEES CENTRALISEE '
RAPPEL SUR LA CONCEPTION D’UN SGBD CENTRALISE
Analyse des besoins PERSONNE_S : SELECT * FROM PERSONNE WHERE SECTEUR = "S"
-> Modèle conceptuel de données
-> Modèle relationnel (logique)
-> Implémentation (physique)
CONCEPTION D’UN SGBD DISTRIBUE
Analyse des besoins
-> Modèle conceptuel de données
-> Modèle relationnel (logique) RECONSTRUCTION PAR UNION
-> Implémentation (physique)
Personne = personne_N UNION Personne_S
(Construction du schéma global)
SQL :
-> Fragmentation/ Réplication / Migration (physique)
SELECT * FROM personne_N
-> Allocation (Placement)
UNION
(Distribution des données)
SELECT * FROM Personne_S
1
FRAGMENTATION VERTICALE - PROJECTION (SELECT) ORCHESTRATION D’UNE JOINTURE INTER-SITE :
Principe de base :
Contrainte : la clé primaire doit appartenir à chaque fragment
- Transférer la plus petite des deux relations (si on n’a pas le choix)
Table :
Solutions possibles
- Algorithme des semi-jointures
- Jointures parallèles
- Jointure parallèle avec partitionnement des données sur attribut de
jointure
- Jointure parallèle sans partitionnement des données sur attribut de
jointure
Algorithme des semi-jointures :
- Réduire la quantité d’information transférée pour exécuter une
jointure quand peu de n-uplets participent à la jointure
Jointures parallèles avec partitionnement des données :
P ER S ON N E_ I D _ N OM _ P R E N OM = S E LE C T I D , N OM , P R E N OM F R O M P E R S O N NE - Distribuer le traitement d’une jointure sur plusieurs machines, en
P ER S ON N E_ I D _ S EC T EU R = S E L EC T I D , S EC T E UR F R OM P ER S O N N E tenant compte du partitionnement
Personne_ID_NOM_PRENOM Personne_ID_SECTEUR Jointures parallèles sans partitionnement des données :
- Eviter de rassembler l’ensemble des fragments horizontaux sur un
même nœud.
ALGORITHMES DE JOINTURES : RAPPELS
Approche par boucle (LOOP)
Pour chaque tuple de R, on parcourt S (avec ou sans index).
Approche par tri-fusion (SORT-MERGE)
- Algorithme du sort-merge join :
- On trie R et S sur les attributs de jointure, puis scan en parallèle pour
permettre un seul parcours de chaque relation
- Partitionnement des données en amont du tri
- Facilite la parallélisation
- Généralement, nécessite la matérialisation les fragments ➔ latence
supplémentaire
- Plusieurs niveaux de fusion en aval du tri
Approche par hachage (HASH)
RECONSTRUCTION PAR JOINTURE
- Algorithme du hash join
Personne = Personne_ID_NOM_PRENOM JOIN Personne_ID_SECTEUR - Transfert de la plus petite des relations dans une structure de hachage
SQL : avec les attributs de jointure comme clé, pour accéder efficacement
SELECT * FROM Personne_ID_NOM_PRENOM PINP aux valeurs
JOIN Personne_ID_SECTEUR PIS ON [Link] = [Link] - Partitionnement des données en amont
ATENTION - Facilite la parallélisation
Sans la clé primaire dans les fragment vertical, reconstruction impossible ! - Création de plusieurs structures de hachage
FRAGMENTATION MIXTE REPLICATION LOGIQUE
L'association des deux !! Solution pour éviter des transferts est la réplication logique de certaines
Exemple : Fragmentation horizontale suivie d’une fragmentation verticale données
de chaque fragment - Rapprocher les données des besoins
Important :
LA FRAGMENTATION EST CORRECTE SI ET SEULEMENT SI ELLE
- Prendre en compte a fréquence de mise à jour des données répliquées
EST :
- Ce qui peut aussi impacter les performances
- Complète (chaque élément doit se trouver dans un fragment) Au niveau configuration :
- Reconstructible (on doit pouvoir recomposer la base à partir de ses - Le nombre de connexions simultanées ➔ max_connections
fragments) - La taille des buffers de manipulation des données partagés ➔ shared_buffers
- Disjointe (chaque élément de la base (hormis les clés) ne doit pas être - La taille des buffers de maintien de l’intégrité des données ➔ wal_buffers
dupliqué) - La taille du cache ➔ effective_cache_size
- La mémoire utilisable pour le traitement d’opérateurs complexes ➔ work_mem
GESTION DE GRANDES MASSES DE DONNEES - CM2 - La mémoire utilisable pour les processus de maintenance ➔ maintenance_work_mem
- Le coût des accès disques
TRAITEMENT DE JOINTURES INTER-SITES Au niveau des données :
Jointures inter-sites : - L’exploitation des contraintes d’intégrités
C’est une jointure interrogeant des données distribuées sur différents - Le partitionnement des tables
sites, le coût de transfert dépendant du plan d’exécution ! Au niveau des requêtes
Un plan d’exécution possible : - L’utilisation d’index
- Celui qui demande, récupère tout et c’est lui qui fait les calculs. - La parallélisation des calculs
- Collaborative - La matérialisation de résultats (intermédiaires)
COMMENT AMELIORER LES PERFORMANCES DE TRAITEMENT ? PARTITIONNEMENT ET PARALLELISATION : LE ‘SHARDING’
Techniques d’exécution distribuée : Principe :
En tenant compte des propriétés du réseau : - Fragmentation horizontale garantissant un partitionnement des
- Row blocking : transfert de données par batch plutôt données pour un traitement en parallèle des fragments
qu’individuellement Choix de fragmentation :
- Multicast : dans le cas de transfert des mêmes données sur plusieurs - Basé sur des plages de valeur de la clé de ‘sharding’ (Range-based
sites, valoriser les chemins les moins coûteux partionning)
- Multithreading : parallélisation de certains opérateurs - Basé sur une valeur de hachage de la clé de ‘sharding’ (Hash
- Problème de communication synchronisée inter-thread partionning)

2
Mise en place du ‘sharding’ en deux étapes : L’évolution de son débit dans le temps :
- Allocation des Shards aux machines (Plusieurs Shards possibles par - Débit constant
machine) - Débit borné
- Affectation des tuples à leur Shard - Débit variant
Range-based partionning : - Avec motifs régulier
EX : Shard 1 : Valeur < 15 - De manière erratique
Shard 2 : 15 < Valeur < 25
Shard 3 : 25 < Valeur
Hash partitionning :
EX : shard 1 : hash( id ) % 3 -> 0
shard 2 : hash( id ) % 3 -> 1
shard 3 : hash( id ) % 3 -> 2
AVANTAGES DU SHARDING
Gain de temps de traitement :
- Traitement en parallèle (sur différentes machine CPU, RAM) des
‘shard’ ➔ Possibilités d’adapter la voilure de l’infrastructure (nombre
de CPU et RAM) en fonction de la quantité de données à traiter
Evite la centralisation des données sur un seul et unique serveur (sécurité)
INCONVENIENTS DU SHARDING
Une mauvaise stratégie de ‘sharding’ (choix de la clé de sharding) peut
rendre contreproductif le processus à cause de ‘shard’ de taille trop
hétérogène
- Rééquilibrage complexe
Sensible à la performance des réseaux REQUETES CONTINUES
Sensible à la disponibilité et l’évolution du cluster de machine
DEFINITION
GESTION DE GRANDES MASSES DE DONNEES - REVISION CM3 Une requête continue est une requête qui est émise une fois et qui
SYSTEME DE GESTION DES FLUX DE DONNEES (DATA STREAM s'exécute de façon logique sur les données jusqu'à ce qu’elle soit
MANAGEMENT SYSTEM DSMS) terminée.
Le traitement d’une requête continue nécessite l’exécution d’opérateurs
Traitement de flux de données à débit variant (avec ou sans état) avec une fréquence et une portée paramétrée.
Contraintes :
OPERATEURS
- Maîtriser la qualité des résultats retournés
- Maîtriser les ressources (CPU, RAM bande passante) nécessaires au Le traitement d’une requête continue peut nécessiter des opérateurs de
traitement différents types
Opérateur sans état (‘Stateless’) :
- Calcul indépendant de l’état/résultat calculé précédemment
- Exemple : requête appliquant un filtre sur les données
Opérateur avec état (‘Stateful’) :
- Calcul dépendant de l’état/résultat calculé précédemment
- Exemple : requête calculant l’évolution du nombre moyen de valeurs
apparaissant toutes les x secondes
REQUETES CONTINUES : PARAMETRES
Exemple de requete : Donner la moyenne sur les 15 dernières minutes du
niveau de polluants HAP observées à partir des 10 000 capteurs répartis
dans la région Auvergne-RhôneAlpes et ce toutes les 5 minutes.
Une requête continue dispose :
- D'une fréquence de résultat attendu : "toutes les 5 minutes."
- D'une portée sur laquelle la requête est calculée : "sur les 15 dernières
minutes."
FENETRAGE
FLUX EN ENTREE Rappel :
- Un flux est infini
DEFINITION
Conséquence :
Un flux S est un ensemble potentiellement infini de paires < s, τ >, où : - Il n’est pas possible d’interroger le flux dans sa globalité
- s est un tuple respectant le schéma de S Besoin :
- τ est le timestamp de cet élément. - Système de fenêtrage
Remarque : Paramètres : taille, pas
- Un timestamp τ n’appartient pas au schéma de S et il peut y avoir Différents types :
zéro, un ou plusieurs éléments de S partageant le même τ. - Fenêtres basées sur le temps
CARACTERISTIQUES - Fenêtres basées sur le nombre de tuples reçus
Un flux S est caractérisé par : - Fenêtres basées sur le nombre de valeurs de tuples reçus
- La valeur de son débit exprimé en nombre de tuples par unité de FENETRES SAUTANTES
temps - Fenêtre glissante où pas = taille

3
FENETRE BASEE SUR LE TEMPS (TIME-BASED) DISTRIBUEE
- La fenêtre est définie à partir d’une durée Grille, Cloud :
- Le calcul de la requête se fait sur les tuples reçus depuis δ secondes et - Possibilité de parallélisation des threads
ce toutes les π secondes - Mise à disposition de nouvelles ressources
- Introduction de latence réseau
SYSTEMES DE GESTION DE FLUX (DSMS)
PERFORMANCE DE TRAITEMENT
Pourvoir obtenir des résultats en quasi-temps réel
Optimiser le traitement de la requête
QUALITE DES RESULTATS
Obtenir des résultats en ayant limité les pertes de données
Eviter la congestion des opérateurs
ADAPTABILITE DES RESSOURCES
Avoir les ressources nécessaires pour traiter le flux
Ne pas bloquer des ressources inutilement
REMARQUE
La fin d’une fenêtre n’implique pas forcément la disponibilité d’un
FENETRE SAUTANTE résultat.
- La fenêtre est définie à partir d’une durée
- Le calcul de la requête se fait sur les tuples reçus depuis δ secondes et
ce, toutes les π secondes

CONGESTION
Le système est considéré comme congestionné, si :
- Des données en entrée du système sont perdues (file d’attente
saturée)
FENETRES SUR LE NB DE TUPLES REÇUS (TUPLE-BASED - Des données avec TTL dépassé
- La fenêtre est définie à partir d’un nombre de tuples reçus ELASTICITE DES DSMS
- Le calcul de la requête se fait sur les n tuples reçus DEFINITION
Adapter les ressources nécessaires au traitement des requêtes
ENJEUX (INTERSECTION BIG DATA / GREEN IT)
Performance du traitement de la requête
Qualité des résultats retournés
Consommation énergétique :
- Pouvoir adapter la quantité de ressources nécessaires en cas de
variations du débit des flux à traiter
Automaticité :
FENETRES SUR LE NB DE TUPLES REÇUS (PARTITION-BASED) - Pouvoir adapter automatiquement les ressources
- La fenêtre est définie à partir d’un nombre de valeurs de tuples reçus GESTION DE L’ELASTICITE
- Le calcul de la requête se fait sur les n tuples reçus et vérifiant une Deux niveaux de traitements de l’élasticité :
formule logique - Ajout/suppression de ressources supplémentaires pour l’exécution
des threads
- Concentration/Etalement des threads
Deux contextes :
- Ressources disponibles et suffisantes
- Ressources limitées et insuffisantes
APPROCHES WORKFLOW
DEFINITION
GESTION DU TEMPS Les requêtes sont considérées comme des graphes d’opérateurs
Problème de synchronisation présent en cas de plusieurs sources (ordre TRAITEMENT DE REQUETES
non garanti, latence réseau…). Choix de la topologie
Système en ‘battement de cœur’ possible de deux manières : - Définition d’une topologie réduisant le nombre de tuples manipulés
- Associer un timestamp à chaque tuple entrant. Choix du degré de parallélisme des opérateurs
- Chaque source envoie une ponctuation lorsqu’elle a fini d’envoyer des - Définition du nombre de threads permettant de paralléliser
tuples associées à un timestamp et la synchronisation s’effectue grâce l’exécution d’un opérateur
aux sources Choix de l’allocation d’une tache sur une unité de traitement
INFRASTRUCTURE - Définition d’une stratégie d’allocation des threads sur les unités de
traitement
CENTRALISEE Choix de l’implémentation des opérateurs
Multicœurs : - Définition d’une stratégie de sélection d’implémentation des
- Possibilité de parallélisation des threads opérateurs
4
STRATEGIE DE REPLICATION TP 2 - GGMD
Réplication incrémentale : DEFINITION DES CONNEXIONS INTER-BASES
- Ajout / suppression d’une réplique à la fois par reconfiguration
- Peut générer de nombreuses reconfigurations Création d’un index sur personne :
Réplication multiple : CREATE INDEX Q2 ON personne (nomprenom,
- Ajout / suppression de plusieurs répliques à la fois par reconfiguration datenaiss, lieunaiss);
- Peut générer de fortes variations de ressources TP 3 - SPARK
STRATEGIES D’ALLOCATION - DIFFERENTES STRATEGIES : LECTURE DE DONNEES
Orienté répartition : - [Link](path): l'appel à cette méthode crée un RDD à
- Round robin des affectations des threads sur les unités de traitements
partir des lignes du fichier. A noter que le path du fichier peut être
- Permet d’équilibrer la charge (si les temps de traitement des tuples
un chemin réseau, ce qui nous sera utile par la suite.
considérés comme identiques)
- Pas de considération du réseau TRANSFORMATIONS
Orienté échange réseau : - map(func): applique une fonction à chacune des données.
- Tient compte de la position des opérateurs dans la topologie pour - filter(func): permet d'éliminer certaines données.
choisir leur affectation - flatMap(func): tout comme map, mais chacune des données
- Permet de réduit le trafic réseau d'entrée peut être transformée en plusieurs données de sortie.
- Mais concentre la charge sur certaines unités de traitement - sample(withReplacement, fraction, seed):
Orienté ressources : récolte un échantillon aléatoire des données. Cette méthode est utile
- Tient compte des ressources (CPU, RAM) disponibles sur une unité de pour tester un algorithme sur un petit pourcentage de données.
traitement L'argument seed permet de réaliser des expériences
- Permet une occupation maximale d’un sous-ensemble minimal de reproductibles. Exemple :.sample(False, 0.01, 42).
ressources (algo sac à dos) - distinct(): supprime les doublons.
- Mais reste faible robustesse en cas de variation du flux - reduceByKey(func): applique une fonction de réduction aux
FOCUS SUR LE DEGRE DE PARALLELISME valeurs de chaque clé. Par exemple, la fonction func peut renvoyer
Changer le degré de parallélisme d’un opérateur. Quand ? : le maximum entre deux valeurs.
- La détection de la congestion de l’opérateur (approche curative) - sortByKey(ascending): utilisée pour trier le résultat par clé.
- La détection d’un risque de congestion de l’opérateur (approche - join(rdd): permet de réaliser une jointure, ce qui a le même sens
préventive) que dans les bases de données relationnelles.
Comment ? ACTIONS
- En répliquant un opérateur (scale out) / supprimant une réplique
(scale in) - reduce(func): applique une réduction à l'ensemble des
données.
TP1 - DEPLOIEMENT D'UNE BASE DE DONNEES REPARTIE - collect(): retourne toutes les données contenues dans le RDD
DEFINITION DES CONNEXIONS INTER-BASES sous la forme de liste.
- count(): retourne le nombre de données contenues dans RDD.
Définir le 'wrapper' qui permettra de vous connecter à la base :
CREATE EXTENSION IF NOT EXISTS postgres_fdw;
CREATE SERVER small FOREIGN DATA WRAPPER EXEMPLE
postgres_fdw OPTIONS (host '[Link]',
port '5432', dbname 'insee'); cleanedTortoises = [Link](lambda t:
Définir l'association utilisateur local/utilisateur distant qui vous len(t) > 1 and int(c[t]) == 21)
donnera le droit d'accès à la base insee :
CREATE USER MAPPING FOR "postgres" SERVER small jsonTortoises = [Link](lambda t: {
OPTIONS( user 'etum2' ,password 'etum2'); "id": int(t[1]), "top": int(t[2]), "position":
Définir les 'foreign tables' dbase insees depuis une VM : int(t[3]), "nbAvant": int(t[4]), "nbTour":
CREATE FOREIGN TABLE remote_personnes(idp int(t[5]), "distanceTotal": 254 * int(t[5]) +
serial4, nom varchar(80), prenoms varchar(80)) int(t[3]) })
SERVER small OPTIONS(schema_name 'public',
min = [Link]( (a, b) => a min b)
table_name 'personnes');
Définir une 'table' à partir d’une 'foreign tables' :
CREATE TABLE personnes AS SELECT * (ou les
colonnes qui nous intéressent si fragmentation
vertical) FROM remote_personnes WHERE (si ya
besoin partition horizontale);
Créer une 'publication' :
CREATE PUBLICATION pub_update_mairie FOR TABLE
mairie;
Créer une 'suscriptions' :
CREATE SUBSCRIPTION sub_update_mairie CONNECTION
'host=[Link] port=5432 password=etum2
user=etum2 dbname=insee' PUBLICATION
pub_update_mairie;
Créer un 'Trigger' sur les updates :
CREATE TRIGGGER TT AFTER UPDATE ON commune c FOR
EACH ROW BEGIN UPDATE personne set lieunaiss =
[Link] where lieunaiss = [Link]; END