Plan

●
Motivation
●
Système distribué
●
Architecture
●
Réplication
●
Fragmentation
●
Optimisation de requêtes
●
Transactions distribuées
●
Systèmes

34 OlivierCuré [ocure@[Link]]
Optimisation des requêtes
●
Dans un SGBD centralisé, le principale critère de pour
mesurer le coût d’un plan d’exécution est le nombre
d’accès disque
●
Dans un SGBD distribué, il faut également prendre en
compte :
●
Le choix de la meilleure localisation des fragments,
●
les coûts liés aux transferts de données, exploitation des
CPU, entrée/sortie.
● l’exploitation de parallélisme
● les transferts entre sites.
●
Il faut étudier les différentes solutions et adopter la plus
économique en fonction de l’utilisation de réplication
et/ou fragmentation.

bases de données distribuées 35

 Parallélisme inter et intra
requête
Inter-requête:
Différents requêtes sont exécutées de manière
concurrente
Augmentation du débit et réduction de la latence
Intra-query
Exécution des opérations d’une simple requête en
parallèle
Diminution de la latence pour des requêtes dont
l’exécution peut être longue

36 OlivierCuré [ocure@[Link]]
Transformation d’un
requête
●
Soit la relation F (fournisseur) et la
requête “Fournir l’ensemble des
tuples de F”.
●
Réplication sans fragmentation
●
Il faut choisir la réplique avec le
moindre coût => site de la requête,
sites des repliques
●
Réplication et fragmentation
●
Implication de jointures et d’unions

bases de données distribuées 37

Transf. réplication et
fragmention vert.
●
F sur 2 sites (A et B), Q = “Obtenir les
fournisseurs français”
σpays=fra f

● Avec f = fA ∪ fB

● Donc: σpays=fra (fA ∪ fB)

●
Qui correspond à
σpays=fra (fA)∪ σpays=fra (fB)

bases de données distribuées 38

Transf. réplication et
fragmention vert. (2)
●
Avec des méta données indiquant que
les fournisseurs français ne sont que
sur le site 1, on peut simplifier en
σpays=fra (fA)

bases de données distribuées 39

Requêtes avec jointures
●
Le choix d’une bonne stratégie
d’exécution des jointures est un point
clé de l’optimisation de requêtes dans
un SGBD centralisé ou distribué
●
Soit 3 relations R1, Ré et R3 sur 3
sites, resp. S1, S2 et S3
●
Une requête sur S1 impose
R1⋈ R2 ⋈ R3

bases de données distribuées 40

Requêtes avec jointures
●
On a le choix entre plusieurs
stratégies, par exemple
●
Transférer des copies des relations
R2 et R3 vers S1 pour tout exécuter
sur S1
●
Transférer R3 sur S2, y exécuter la
jointure et transférer ce résultat sur
S1 pour y faire la jointre avec R1
●
Transférer R1 sur S2, y exécuter la
jointure, transférer le résultat sur S3,
y faire la jointure et transférer le
résultat final sur S1
●
.. bases de données distribuées 41
Requêtes avec jointures
●
Pas une stratégie meilleure qu’une
autre dans le cadre général
●
Cela dépend de la taille des relations
à transférer, le coût de la transmission
et la vitesse de traitement sur chaque
site (par exemple, accès à des
indexes)

bases de données distribuées 42

Requêtes avec jointures
●
Exemple: transfert de R2 et R3 sur S1.
La requête gagne à exploiter les
indexes de R2 et R3
●
Re-création des indexes de R2 et R3
sur S1?
●
Transmission des indexes sur S1?
●
Exploitation d’une autre méthode de
jointure?

bases de données distribuées 43

Stratégie semijoin
●
On suppose la jointure r1⋈ r2, avec r1
sur S1 et r2 sur S2. Ri est le schéma
de ri
●
On veut le résultat sur S1
●
Si on a un grand nombre de tuples de
r2 qui vont pas joindre avec r1
(dangling tuples) alors le transfert de
r2 vers S1 est coûteux et inutile

bases de données distribuées 44

Stratégie semijoin
●
Une stratégie possible:
1. Temp1 = ∏R1∩R2 (r1)

2. Transfert Temp1 sur S2

3. Temp2 = r2 ⋈Temp1 sur S2
4. Transfert temp2 sur S1
5. Calcul de r1 ⋈Temp2 sur S1

bases de données distribuées 45

Stratégie semijoin
●
Vérification
r1 ⋈ r2 ⋈ ∏R1∩R2 (r1)

avec r1 = r1 ⋈ ∏R1∩R2 (r1)

donc
r1 ⋈ r2 ⋈ ∏R1∩R2 (r1) = r1 ⋈ r2

bases de données distribuées 46

Stratégie semijoin
●
Cette stratégie est avantageuse
quand elle permet d’envoyer moins de
tuples de r2 sur S1.
●
La semijointure de r1 avec r2
(dénotée r1⋉ r2) correspond à
∏R1 (r1 ⋈ r2)

bases de données distribuées 47

Exemple(1/4)

Soit la BDR composée des relations suivantes :

P (NP, NOMP, MADE_IN, COULEUR, POIDS)
U (NU, VILLEU, NOMU)
F (NF, NOMF, VILLEF, ADRESSE, PAYS, COEF)
PUF (NP, NU, NF, DATE, QUANTITE)
Les relations U et PUF sont sur le site A.
Les relations F et P sont sur le site B.
Le réseau reliant les deux sites A et B a les caractéristiques
techniques suivantes :
temps d'accès d'un site à un autre : 0,5 seconde.
débit de transmission : environ 1 000 octets d'informations
utiles / secondes

bases de données distribuées 48

Exemple(2/4)

Transfert d'une message en :

( 0,5 + Nb-d'octets-du-message /1000) secondes.

A la requête "Donner les numéros et les noms des

fournisseurs qui ont livré un produit italien à une usine
située à Lausanne", qui provient du site B, correspond la
requête algébrique suivante :
R = (NOMF, NF) ((VILLEU = "Lausanne")U ⋈ PUF ⋈
(MADE_IN = "Italy")P ⋈ F)

bases de données distribuées 49

Exemple (3/4)
●
Soient les deux stratégies d'exécution suivantes :
●
S1 : Sélectionner les produits italiens sur B. Pour chacun de ces
enregistrements, interroger A pour savoir si le produit a été livré
dans une usine de Lausanne et par qui. La réponse contient les
numéros (NF) des fournisseurs qui livrent ce produit à Lausanne.
●
S2 : Sélectionner les numéros (NP) des produits italiens sur B.
Transmettre ces numéros au site A en lui demandant qu'il renvoie
les numéros des fournisseurs qui ont livré un de ces produits à
une usine de Lausanne. Le site A transmet un ensemble de
numéros de fournisseurs à B.

bases de données distribuées 50

Exemple (4/4)
F contient 100 fournisseurs. P contient 500 produits italiens, qui
concernent 800 livraisons à Lausanne. Les numéros (NP et NF) sont
codés sur 2 octets.

Avec la première stratégie, 500 messages contenant un numéro de

produit sont envoyés de B vers A et autant de réponses (contenant
800 numéro de fournisseurs, dont beaucoup de doubles) sont
renvoyées. Le temps de transfert de données est donc au total :

T1 = 500 * (0,5 + 2 / 1000) + 500 * 0,5 + 800 * 2 / 1000 = 503

secondes (8 minutes)

Avec la stratégie S2, un message contenant 500 numéros est envoyé

au site A, qui renvoie à B un message contenant au plus 100
numéros de fournisseurs (puisqu'il n'y a que 100 fournisseurs dans
la BD!). Le temps de transfert de données est donc au total :

T2 = 0,5 + (500 * 2) / 1000 + 0,5 + (100 * 2) / 1000 soit 1,7 seconde

bases de données distribuées 51

 Plan
●
Motivation
●
Système distribué
●
Architecture
●
Réplication
●
Fragmentation
●
Optimisation de requêtes
●
Transactions distribuées
●
Systèmes

52 OlivierCuré [ocure@[Link]]
Transactions distribuées
●
Deux constats :
●
les utilisateurs modifient les bases de données
●
il y a des pannes du matériel, des logiciels, du
réseau
●
Comment assurer l'intégrité des
données
●
concept de transaction
●
Une transaction consiste en l'exécution d'un
programme.

bases de données distribuées 53

Transactions ACID

A : Atomique (exécution en un bloc ou rien)

C : Cohérence (d'un état cohérent à un autre
état cohérent de la BD).
I : Isolation (màj non visibles des autres
transactions tant qu'il n'y a pas de commit).
D : Durabilité des modifications sur la BD
(persistence des txn commitées)

bases de données distribuées 54

Opérations élémentaires
d'une transaction
●
Les opérations d'une transaction sont
contenues entre "begin of
transaction" et "end of transaction".
●
Les opérations élémentaires sont :
●
read
●
write
●
commit, qui valide et enregistre le nouvel état
de la bd.
●
abort, abandon et retour à l'état initial de la
bd.

bases de données distribuées 55

Transactions et BDD
●
Distinction entre txn
●
Locale
●
Elle accède à des données situées sur
le site initiant la txn
●
Globale
●
Elle accède à des données situées sur
une site différent de celui initiant la txn
ou accède à des données situées sur
plusieurs sites
●
Implantation plus complexe

bases de données distribuées 56

Architecture standard
●
Comme dans un SGBD centralisé, il y a un txn
manager (TM) qui gère l’exécution des txn
sur son site. Chaque site a son txn manager
●
En plus, dans un SGBDD, chaque site possède un
txn coordinator (TC) qui coordonne
l’exécution des différentes txn initiées sur son
site
Machine 1 Machine n

TC1 TC1
TCn

TM1 TMn

bases de données distribuées 57

Txn manager
●
Il maintient un journal des logs pour permettre la
reprise en cas de panne
●
Participe au contrôle de la concurrence pour
coordonner l’exécution concurrente de txn
s’exécutant sur son site.

bases de données distribuées 58

Txn coordinator
●
Il est responsable
●
du démarrage de l’exécution d’une txn
●
du découpage d’une txn en sous-txn et à
la distribution de ces sous-txn aux sites
appropriés
●
de la coordination de la terminaison
d’une txn, qui peut se conclure par un
commit ou un abort sur l’ensemble des
sites impliqués

bases de données distribuées 59

Pannes possibles
●
Comme pour un système centralisé, on peut
rencontrer des problèmes logiciel, matériel,
crash disque, etc.:
●
Problèmes supplémentaires d’un SGBDD
●
Panne d’un site
●
Perte de messages
●
Panne du lien de communication entre sites
(reroutage)
●
Partitionnement du réseau

bases de données distribuées 60

Txn globale
●
Risque d’une txn globale
●
Opérations sont commitées sur un
site et pas sur un autre → état
incohérence
●
Il faut garantir l’atomicité même en
cas de txn distribuée
●
Pour cela, on utilise un protocole de
validation (commit protocol)
●
Parmi les plus simples et le plus
utilisé, on retrouve le protocole de
validation à 2 étapes (2 phase
commit = 2PC)
bases de données distribuées 61
2PC
● Soit une txn T, initiée sur Si où le txn
coordinateur est Ci
●
Lorsque T a complété son exécution
(chaque site impliqué par T a géré
sa part de la txn et à informé Ci )
alors Ci débute la protocole 2PC

bases de données distribuées 62

2PC
●
Phase 1:
● C ajoute <prepare T> au journal et force
i
la persistence du journal
● C envoie <prepare T> à chaque site
i
ayant exécuté un morceau de T
●
A la réception de ce message sur chaque
site, le txn manager détermine s’il est
possible de valider la txn locale.
●
Soit il ajoute <no T> à son journal
et envoie un abort T à Ci
●
Soit il ajoute <ready T> à son
journal, persiste le journal et envoie
<ready T> à Ci
bases de données distribuées 63
2PC
●
Phase 2:
● C réceptionne les réponses à son
i
message <prepare T> depuis l’ensemble
de sites impliqués par T (après un laps de
temps, il peut considérer que T doit être
annulée)
●
T peut être validée (commit) si tous les
sites ont envoyé un <ready T>, Ci envoie
alors un <commit T> aux sites
● Sinon il faut annuler T et C envoie un
i
<abort T> aux sites
●
A la réception des messages, les sites
enregistrent la décision dans le log et
applique l’opération
bases de données distribuées 64
2PC
●
Phase 2:
●
Dans certaines implémentations, les sites
envoient un <acknowledge T> à Ci .
Celui-ci ajoute alors <complete T> au
journal

bases de données distribuées 65

2PC et panne d’un site
●
Si un site “tombe” avant l’envoi d’un <ready T>,
Ci considère cela comme un <abort T>
●
Si un site “tombe” après l’envoi de son <ready
T>, Ci exécute le reste du protocole de
validation en ignorant la panne
●
Lorsqu’un site redémarre après une panne, il doit
examiner son journal pour déterminer l’état
des txn qui étaient bloquées localement

bases de données distribuées 66

2PC et panne d’un site
●
Plusieurs situations à l’analyse du journal suivant
que le site y trouve
●
Un <commit T> alors il valide T
●
Un <abort T> alors il annule T
●
Un <ready T>, le site doit alors consulter
Ci. Si il fonctionne, il va notifier le site de
la décision sur T. Sinon (e.g., Ci est
injoignable), le site va envoyer des
<queryStatus T> aux autres sites du
système qui vont chercher dans leur
journal. Si le site n’a pas de réponse, il
repoussera la prise de décision sur T et
enverra régulièrement des messages aux
autres sites
bases de données distribuées 67
2PC et panne d’un site
●
Le log ne contient aucune information sur
T (donc le site est tombé avant le
<prepare T>). Il doit annuler

bases de données distribuées 68

2PC et panne du txn coord
● Si Ci “tombe”, les sites impliqués par T doivent
prendre une décision. Cela n’est pas toujours
possible et dans ce cas, ils doivent attendre un
redémarrage du site de Ci (pb de blocage).
●
La décision est possible si
●
Un site a un <commit T> alors on valide
sur les sites
●
Un site a un <abort T> alors on annule
●
Un site n’a pas de <ready T>. Cela veut
dire que Ci n’a pas été capable de
décider. Dans ce cas, il est préférable
d’annuler T
bases de données distribuées 69
2PC et partitionnement du
réseau
● Ci et les sites de T sont dans la même partition du
réseau : le partitionnement n’a pas d’effet sur le
protocole
● Ci et les sites se retrouvent sur différentes partitions
du réseau. Les sites se trouvant dans la partition de
Ci suivent une exécution normale du protocole

srv1

srv3

srv2

bases de données distribuées 70

3PC
●
Il existe des alternatives à 2PC,
comme 3PC
●
3PC élimine certains désavantages de
2PC (principalement le pb de
blocage) mais au prix d’une
complexité et d’une dégration des
performances importantes

bases de données distribuées 71

Contrôle de la
concurrence
●
Les mécanismes de contrôle de la concurrence
d’un SGBD centralisé peuvent être adaptés à
un environnement distribué
●
On considère que les modifications doivent être
faites sur l’ensemble des répliques d’une
donnée
●
Dans ce cas, si un site n’est pas capable
d’effectuer une modification localement alors
les modifications ne peuvent être traitées

bases de données distribuées 72

Protocoles basés sur les
verrous
●
Approche avec un unique gestionnaire de verrou
● Il se trouve sur le site S
i
●
Toutes les demandes de verrous ou
libération de verrous se font en envoyant
un message à Si
●
Avantages: implantation simple, unique
gestionnaire de détection de verrou
mortel (deadlock)
● Inconvénients: Si est un goulet
d’étranglement, vulnérabilité (SPOF)

bases de données distribuées 73

Protocoles basés sur les
verrous
●
Approche avec une distribution de la
gestionnaire de verrous
●
Chaque site à son lock manager local qui
gère ses données locales
●
Avantages: Réduction l’effet goulet
d’étranglement
●
Inconvénients: gestion des deadlock est
plus complexe qu’avec un unique
gestionnaire puisque lock et unlock sont
distribués. On peut avoir des deadlock
intersites sans avoir de deadlock local

bases de données distribuées 74

Protocoles basés sur les
verrous
●
Approche avec une copie primaire
●
Une réplique est élue comme primaire
●
Une txn demandera un lock/unlock à
cette réplique
●
Avantages: implantation simple
●
Inconvénients: SPOF

bases de données distribuées 75

Protocoles basés sur les
verrous
●
Approche avec protocole de majorité
●
N répliques, il faut joindre plus de la
moitié des répliques pour obtenir un
verrou. Chaque réplique(site) répondra si
verrou possible ou pas.
●
La txn ne s’exécute pas tant que la
moitié des sites ont accordé le verrou
●
Avantages: bonne gestion des pannes de
certains sites
●
Inconvénients: implantation, gestion des
deadlocks

bases de données distribuées 76

Protocoles basés sur
l’horodatage
●
On attribue un unique timestamp à chaque txn
pour permettre au système de décider l’ordre de
sérialization
●
Deux méthodes pour générer des timestamps
uniques:
●
Centralisée: une unique site va fournir les
timestamps. En se basant sur son horloge
local ou un compteur par exemple
●
Distribuée: chaque site génère d’uniques
timestamps (compteur local ou horloge).
Timestamp = concaténation de l’identifiant
du site et timestamp local

bases de données distribuées 77

Protocoles basés sur
l’horodatage
●
Problèmes possibles
●
Solution basée sur des compteurs
●
Le compteur d’une machine
augmente plus rapidement sur
une machine que les autres
●
Solution basée sur des horloges
●
Synchronisation des horloges
des différentes machines

bases de données distribuées 78

Administration
●
SGBD centralisé: un DBA a le contrôle
la BD
●
SGBD distribué: 1 DBA global
(responsable du système) + DBA
locaux (sur chaque site) avec une
possibilité d’autonomie

bases de données distribuées 79

 Plan
●
Motivation
●
Système distribué
●
Architecture
●
Réplication
●
Fragmentation
●
Optimisation de requêtes
●
Transactions distribuées
●
Systèmes

80 OlivierCuré [ocure@[Link]]
 Systèmes

●
Fondé en 1979
●
Shared nothing

●
Fondé en 2000, acquis par IBM en 2010
●
SGBDD parallèle Shared nothing
81 OlivierCuré [ocure@[Link]]
 Systèmes

●
Fondé en 2003, acquis en 2010 par EMC
●
SGBDD parallèle Shared nothing

●
Fondé en 2005, acquis par HP en 2008
●
SGBDD parallèle

82 OlivierCuré [ocure@[Link]]
 Systèmes

●
Fondé en 2003, acquis en 2008 par Microsoft
●
SGBDD parallèle Shared nothing

●
Fondé en 2005, acquis par Terradata en 2011
●
SGBDD parallèle Shared nothing, SQL au dessus
de MapReduce

83 OlivierCuré [ocure@[Link]]
 Systèmes

●
Fondé en 2003, acquis par Sun
SGBDD parallèle Shared nothing
Réplication synchrone
2PC
Partitionnement horizontal: hashage basé sur la clé
primaire de la table (mais configurable)

84 OlivierCuré [ocure@[Link]]
 Systèmes

●
Création en 1996
Plusieurs solutions
PGCluster II: shared disk
Citus (1 version gratuite et 2 autres $$): réplciation
sync et async
Postgres-XL: replication sync

85 OlivierCuré [ocure@[Link]]
 SQL-on-Hadoop players
●
Impala (Cloudera)
●
Stinger (Hortonworks) Hive via ORCfile and Tez
●
Drill (MapR)
●
Hadapt (now Teradata)
●
Apache HAWQ

86 OlivierCuré [ocure@[Link]]