Algorithmes distribues pour le big data @ DevoxxFR 2015

@doanduyhai #AlgosBigData

Algorithmes Distribués pour le BigData

@doanduyhai [email protected] Datastax Évangéliste technique Cassandra


Programme Comptage de grands nombres avec HyperLogLog

Consensus distribué avec Paxos

2

@YourTwitterHandle @YourTwitterHandle @doanduyhai #AlgosBigData

HyperLogLog Philippe Flajolet 2007


Compter le nombre d’éléments distincts, en distribué, dans un ensemble de grande cardinalité

Le défi

4



Le défi

5



Le défi

6


Quelques approches possibles

7

Technique Stockage requis Nb d’éléments estimés Marge d’erreur

Java HashSet 10 447 016 (10M) 67 801 0%

Linear Probabilistic Counter 3 384 (3k) 67 080 1%

HyperLogLog 512 70 002 3% Source: http://highscalability.com/

@doanduyhai #AlgosBigData 8

Jouons un peu !


Lancés de dé

9

0

2

4

6

8

10

12

14

16

18

20

1 2 3 4 5 6

100 lancés de dé


Lancés de dé

10

0

20

40

60

80

100

120

140

160

180

200

1 2 3 4 5 6

103 lancés de dé


Lancés de dé

11

0

20000

40000

60000

80000

100000

120000

140000

160000

180000

1 2 3 4 5 6

106 lancés de dé


Avant HyperLogLog, il y avait … LogLog

Algorithme LogLog

12


Algorithmesimplifié LogLog 1)  Choisir une fonction de hachage H très distributive

2)  Pour chaque élément observé (login, article_id, uuid …), appliquer H

3)  Convertir le hash en séquence binaire

4)  Déduire des séquences binaires, la cardinalité

13

0111010010101… 0010010010001… 1010111001100…

…


Intuition LogLog Équi-probabilité: 50% des séquences commencent par 0xxxxx 50% des séquences commencent par 1xxxxx 1/4 des séquences commencent par 00xxxxx 1/4 des séquences commencent par 01xxxxx 1/4 des séquences commencent par 10xxxxx 1/4 des séquences commencent par 11xxxxx

14


Intuition LogLog

15

000000…0001xxxxxxx

On repère le 1er bit mis à 1 à la position r à partir du début 000000001xxxx à r = 9 0001xxxxxxxxx à r = 4 000001xxxxxxx à r = 6

rang r


Intuition LogLog

16

000000…0001xxxxxxx

Il y a 2r combinaisons de chaine de bits de longueur r

000…0001, 000…0010, 000…0011,…, 111…1111

rang r


Intuition LogLog

17

Équi-probabilité:

1/2r des séquences commencent par 000000…0001xxx 1/2r des séquences commencent par 000000…0010xxx … 1/2r des séquences commencent par 111111…1111xxx


Raisonnons à l’envers !



J’ai autant de chance de voir 000000…0001xxx que de voir 000000…0010xxx

que de voir 000000…0011xxx etc…



Si j’ai observé 000000…0001xxx je verrai probablement 000000…0010xxx

et également 000000…0011xxx etc…



Si j’ai observé 000000…0001xxx il y a probablement 2r séquences binaires de rang r …



Si j’ai observé 000000…0001xxx il y a probablement 2r séquences binaires de rang r …

cardinalité estimée


Formule LogLog Cherchons la position 000…01xxx la plus longue observée parmi tous les séquences binaires

23

cardinalité n ≈ 2max(r)


Problème avec LogLog Exemple: sur 1000 éléments distincts (210 = 1024)

0010000100xxxxxxxxxx 0011001010xxxxxxxxxx 0000000001xxxxxxxxxx … 000000000000001xxxxx à rang r = 15. Cardinalité = 215 = 32768 (FAUX !) … 1100110100xxxxxxxxxx

24


Problème avec LogLog

25

Éléments exceptionels


Approche HyperLogLog 1)  Éliminer et lisser les éléments exceptionnels ☞ moyenne harmonique

26

H =n

1x1+1x2+...+ 1

xnSource: Wikipedia


Approche HyperLogLog Ex: moyenne harmonique de 3, 6, 7, 2 et 120

Moyenne arithmétique = 51 …

27

H =5

13+16+17+112

+1120

≈ 6.80


Approche HyperLogLog 2)  Distribuer le calcul ("diviser pour mieux régner") ☞ appliquer LogLog à n buckets

28

101101000xxxxxxx p = longueur du préfix (ici 6) nombre de buckets = 2p (ici 64)


Approche HyperLogLog 2)  Distribuer le calcul ("diviser pour mieux régner")

29

000000xxx

B1 B2 B3 B32 B33 B34 B64 … …

000001xxx 000010xxx 010000xxx 010001xxx 010011xxx 111111xxx

Flux de données


Approche HyperLogLog 3)  Appliquer LogLog dans chaque bucket

30

101101 0000001xxx p = préfix du bucket

r = rang pour LogLog


Approche HyperLogLog Pour chaque bucket i, on calcule la cardinalité estimée du bucket, Mi

31

Mi ≈ 2max(ri)

ri = rang maximal trouvé dans le bucket Mi


Formule HyperLogLog Moyenne harmonique des Mi, H(Mi), par définition

32

H(Mi) ≈ n/b

n = nb total d’éléments distincts (cardinalité recherchée) b = nb de buckets

☞ n ≈ b・H(Mi)


HyperLogLog, les maths

33

H (xi ) =b

1x1+1x2+...+ 1

xb

= b 11xii=1

b∑

"

#

$$$$

%

&

''''

H (xi ) = b1xi

i=1b

∑"

#$$

%

&''

−1

= b xi−1

i=1

b∑"

#$

%

&'−1


HyperLogLog, les maths On remplace les xi dans la formule par Mi

34

H (Mi ) = b Mi−1

i=1

b∑( )

−1

On remplace les Mi dans la formule par 2max(ri)

H (Mi ) = b 2i−max(ri )

i=1

b∑#

$%

&

'(−1


HyperLogLog, les maths On remplace dans la formule initiale n ≈ b・H(Mi)

35

n ≈αbb2 2−max(ri )

i=1

b∑$

%&

'

()−1

n = cardinalité estimée

b = nb de buckets rang max dans chaque bucket 𝛼b = constante correctrice


Quelles applications ? Nb de visiteurs uniques sur une page web Nb de clicks uniques sur un article en ligne Top N éléments (visiteurs, articles, …) …

36


Exemples d’implémentation Cassandra: estimer taille d’une table distribuée Redis: data structure de base DataFu Pig: UDF standard Twitter Algebird: algorithmes d’algèbre pour Storm & Scalding

37

@YourTwitterHandle @YourTwitterHandle @doanduyhai #AlgosBigData

Paxos Leslie LAMPORT 1989


Trouver un consensus dans un système distribué, en présence de pannes aléatoires(réseau, machine,…).

Le défi

39



Le défi

40



Le défi

41


•  protocole bloquant •  intervention humaine si manager

down

2 phase commit ?


•  état incohérent possible si partition réseau

3 phase commit ?


•  2 allers/retours •  3/4 rôles •  Proposer/Leader •  Acceptor •  Learner

•  A besoin d’un quorum (majorité stricte)

Paxos


Paxos phase 1: prepare

45

Proposer/Leader

Acceptor

prepare(n)

Acceptor

Acceptor

Acceptor

Acceptor

prepare(n)

prepare(n)

n = numéro de séquence

Client


Paxos phase 1: promise

46

Proposer/Leader

Acceptor

Acceptor

Acceptor

Acceptor

Acceptor

promise()

promise()

promise()


Paxos phase 2: accept

47

Proposer/Leader

Acceptor

Acceptor

Acceptor

Acceptor

Acceptor

accept(n,val)

val = valeur du consensus

accept(n,val)

accept(n,val)


Paxos phase 2: accepted

48

Proposer/Leader

Acceptor

Acceptor

Acceptor

Acceptor

Acceptor

accepted(n,val)

accepted(n,val)

accepted(n,val)

Client


Paxos phase 2: learn

49

Proposer/Leader

Acceptor

Acceptor

Acceptor

Acceptor

Acceptor

Learner

Learner val

val

val

Client Learner


Paxos phase 1: prepare

50

Le proposer choisit un nombre n (séquence toujours croissante)

Il envoie prepare(n) à un quorum d’Acceptors

Proposer/Leader prepare(n)

Acceptor



51

Chaque acceptor, à la réception d’un prepare(n):

•  s’il a déjà accepté un accept(m,valm) de la part d’un autre proposer avec m ≤ n ☞ retourne promise(n,(m,valm))

Proposer/Leader

Acceptor

promise(n,(m, valm))



52


•  s’il n’a accepté aucune proposition encore (accept(?,?)) ou s’il a retourné une promesse avec m < n ☞ retourne promise(n, ∅) ET promet de ne plus accepter aucun prepare(m) ou accept(m,?) avec m < n

Proposer/Leader

Acceptor

promise(n,∅)



53


•  s’il a déjà fait une promesse/accepté une valeur avec m > n ☞ ignore la requête. Il peut renvoyer un Nack également (optimisation)

Proposer/Leader

Acceptor

Nack


Paxos phase 1: objectifs

54

Buts de la phase 1:

•  découvrir toute proposition en cours pour la faire progresser

•  bloquer toute ancienne proposition qui n’a pas abouti


Paxos phase 2: accept

55

Le leader reçoit plusieurs promise(n,(mi,vali)):

•  si tous les couples (mi,vali) reçus sont vides (promise(∅, ∅)), le leader peut envoyer accept(n,val) avec val de son choix

•  extrait tous les couples (mi,vali) pour garder vali avec le mi le plus grand ET envoie accept(n,valmax(mi)

) au quorum d’acceptors

Proposer/Leader

Acceptor

accept(n,valmax)


Paxos phase 2: accepted

56

Chaque acceptor à la réception d’un accept(n,val):

•  s’il n’a fait aucune promesse avec m > n, retourne accepted()

•  sinon, ignore la requête

Proposer/Leader

Acceptor

accepted(n,val)


Paxos phase 2: learn

57

Chaque acceptor après avoir envoyé un accepted():

•  envoie la valeur val choisie à une liste de learners (stockage durable)

Le consensus est atteint et a pour valeur val ! Ceci définit un tour de Paxos


Limites du Paxos théorique

58

Une fois la valeur val choisie, on ne peut plus la modifier !

Besoin de faire un reset de val pour un autre tour de Paxos

Multi-Paxos

•  plusieurs tours de Paxos en parallèle •  chaque serveur peut devenir tour à tour Proposer, Acceptor & Learner

Fast-Paxos, Egalitarian-Paxos etc …


Situations de conflits

59

Le dernier arrivé fait progresser une proposition "en cours"

a1

a2

a3

a4

a5

prepare(n1) promise(∅)

promise(∅)

promise(∅)

prepare(n1)

prepare(n1)

Légende message reçu message envoyé

accept(n1,a) prepare(n2)

prepare(n2)

prepare(n2)

promise(n2,(n1,a))

promise(∅)

promise(∅)

propose(n2,a)

propose(n2,a)

propose(n2,a)

accepted()

☠ ☠

accept()

accept()

accept()



60

Annulation d’une promesse précédente

a1

a2

a3

a4

a5


promise(∅)

promise(∅)

prepare(n1)

prepare(n1)


accept(n1,a)

prepare(n2)

prepare(n2)

prepare(n2)

promise(∅)

promise(∅)

promise(∅)

propose(n2,b)

propose(n2,b)

propose(n2,b)

accepted()

accepted()

accept(n1,a)

accept(n1,a)

accepted() ❌

Nack

accepted()

accepted()



61

Inter-deadlock

a1

a2

a3

a4

a5


prepare(n2)

prepare(n2)

prepare(n2)

promise(∅)

promise(∅)

promise(∅)

accept(n2,b)

accept(n2,b)

accept(n1,a)

accept(n2,b) Nack

Nack accept(n1,a)

accept(n1,a) prepare(n1) promise(∅)

promise(∅)

promise(∅)

prepare(n1)

prepare(n1)


promise(∅)

promise(∅)

prepare(n3)

prepare(n3)

…

…

…



62

Inter-deadlock

a1

a2

a3

a4

a5


prepare(n2)

prepare(n2)

prepare(n2)

promise(∅)

promise(∅)

promise(∅)

accept(n2,b)

accept(n2,b)

accept(n1,a)

accept(n2,b) Nack

Nack accept(n1,a)

accept(n1,a) prepare(n1) promise(∅)

promise(∅)

promise(∅)

prepare(n1)

prepare(n1)


promise(∅)

promise(∅)

prepare(n3)

prepare(n3)

…

…

…


Quelles applications ? Élection de master dans les architectures master/slave Atteindre un consensus distribué Créer un algorithme de Compare & Swap distribué

Créer un lock distribué

63


Exemples d’implémentation Cassandra: lightweight transaction Google Chubby/Spanner: lock/transaction distribué(e) Heroku: à travers le framework Doozerd pour gérer la conf Neo4j (≥ 1.9): en replacement de ZooKeeper pour la haute disponibilité

64


CREATE TABLE users( login text PRIMARY KEY, firstname text, … );

Cassandra Lightweight Transaction

65

Contrainte d’unicité


INSERT INTO users(login,firstname,…) VALUES(‘jdupond’,’Jean’,…) IF NOT EXISTS;


66

Contrainte d’unicité


CREATE TABLE locks( lock_type text PRIMARY KEY, lock_id uuid, … );


67

Compare & swap distribué


//Acquérir un lock UPDATE locks USING TTL 60 SET lock_id = 110e8400-e29b-11d4-a716-446655440000 WHERE lock_type = ‘TRANSACTION_ACHAT’ IF lock_id = null; //Relâcher un lock UPDATE locks SET lock_id = null WHERE lock_type = ‘TRANSACTION_ACHAT’ IF lock_id = 110e8400-e29b-11d4-a716-446655440000;


68

Compare & swap distribué


Q & R

! " "

69


MERCI "

70

Technology

Algorithmes distribues pour le big data @ DevoxxFR 2015