Hachage et structures distribuées¶

Les structures de données distribuées sont au cœur de tout système qui passe à l'échelle — le hachage consistant, les filtres probabilistes et les CRDTs en sont les briques fondamentales.

Le problème du hachage classique¶

Le hachage classique assigne chaque clé a un nœud via hash(cle) % N ou N est le nombre de nœuds. Ce schéma fonctionne parfaitement tant que N est fixe. Des qu'un nœud s'ajoute ou disparait, N change, et la quasi-totalité des clés doit migrer vers un nouveau nœud.

Exemple concret : un cluster de 4 nœuds passe a 5. Une clé qui donnait hash(k) % 4 = 2 donne maintenant hash(k) % 5 = 3. La clé est assignee a un nœud différent. Dans un cache distribué, cela signifie un cache miss massif — potentiellement sur 75 a 80% des clés. Dans une base de données shardee, cela déclenche une migration de données catastrophique.

Le hachage consistant résout ce problème en ne remappant qu'une fraction des clés lors d'un changement de topologie.

Hachage consistant : l'anneau de Karger¶

Le hachage consistant, introduit par Karger et al. en 1997, dispose les nœuds et les clés sur un anneau virtuel de \(2^{32}\) positions (ou \(2^{64}\)). Chaque nœud occupe une position determinee par hash(nom_noeud). Chaque clé est assignee au premier nœud rencontre en parcourant l'anneau dans le sens horaire.

graph TD
    subgraph "Anneau de hachage consistant"
        direction LR
        K1["cle-A\n(pos 15)"] --> N1
        K2["cle-B\n(pos 42)"] --> N2
        K3["cle-C\n(pos 78)"] --> N3
        K4["cle-D\n(pos 95)"] --> N1
    end
    N1["Noeud-1\n(pos 10)"]
    N2["Noeud-2\n(pos 40)"]
    N3["Noeud-3\n(pos 75)"]

Quand un nœud N4 rejoint l'anneau en position 60, seules les clés situees entre la position de N3 (75) et celle de N4 (60) — parcourues dans le sens inverse — migrent de N3 vers N4. Toutes les autres clés restent inchangees.

Impact d'un changement : avec K nœuds, l'ajout ou la suppression d'un nœud ne remplacé que \(1/K\) des clés en moyenne. Avec le hachage classique, \((K-1)/K\) des clés sont deplacees.

Nœuds virtuels¶

Le hachage consistant pur souffre d'un problème de distribution : avec peu de nœuds physiques, la répartition des clés est inegale. Un nœud peut se retrouver responsable de 40% de l'anneau pendant qu'un autre n'en couvre que 10%.

La solution est d'introduire des nœuds virtuels (vnodes) : chaque nœud physique est représenté par V positions sur l'anneau, calculees via hash(nom_noeud + "#" + i) pour i de 0 a V-1. Avec V=150, la distribution converge rapidement vers l'uniform par la loi des grands nombres.

Les nœuds virtuels permettent aussi une hétérogénéité maîtrisée : un nœud avec deux fois plus de RAM peut se voir assigner deux fois plus de vnodes, recevant proportionnellement plus de trafic.

Implémentation¶

import hashlib
from bisect import bisect, insort

class ConsistentHash:
    def __init__(self, vnodes=150):
        self.vnodes = vnodes
        self.ring = []          # liste triee des positions
        self.nodes = {}         # position -> noeud physique

    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def add_node(self, node: str):
        for i in range(self.vnodes):
            pos = self._hash(f"{node}#{i}")
            insort(self.ring, pos)
            self.nodes[pos] = node

    def remove_node(self, node: str):
        for i in range(self.vnodes):
            pos = self._hash(f"{node}#{i}")
            self.ring.remove(pos)
            del self.nodes[pos]

    def get_node(self, key: str) -> str:
        if not self.ring:
            raise RuntimeError("Aucun noeud disponible")
        pos = self._hash(key)
        idx = bisect(self.ring, pos) % len(self.ring)
        return self.nodes[self.ring[idx]]

# Usage
ch = ConsistentHash(vnodes=150)
for node in ["node-1", "node-2", "node-3"]:
    ch.add_node(node)

print(ch.get_node("user:42"))    # -> "node-2"
print(ch.get_node("order:999"))  # -> "node-1"

Avant (3 noeuds) :
  user:1   -> node-1
  user:2   -> node-3
  user:3   -> node-2
  user:4   -> node-1

Apres ajout de node-4 :
  user:1   -> node-1   (inchange)
  user:2   -> node-4   (migre depuis node-3)
  user:3   -> node-2   (inchange)
  user:4   -> node-1   (inchange)

Seule user:2 a migre — 25% au lieu de 75% avec hachage classique.

Bloom Filters¶

Un filtre de Bloom répond à la question "est-ce que cet élément appartient a cet ensemble ?" en utilisant une fraction de la mémoire d'un ensemble classique. La contrepartie : il peut répondre "oui" a tort (faux positif), mais ne répond jamais "non" a tort (jamais de faux negatif).

Structure et fonctionnement¶

Un filtre de Bloom est un tableau de m bits initialise à zero, associe a k fonctions de hachage indépendantes. Pour inserer un élément x :

1. Calculer \(h_1(x), h_2(x), \ldots, h_k(x)\) — k positions dans le tableau
2. Mettre les k bits correspondants a 1

Pour tester si x est present :

1. Calculer les k positions
2. Si tous les k bits sont a 1 → "probablement present"
3. Si au moins un bit est a 0 → "certainement absent"

Dimensionnement¶

Les parametres optimaux pour n éléments et un taux de faux positifs p cible :

Avec ces valeurs optimales, le taux de faux positifs effectif est :

import math

def bloom_params(n: int, p: float) -> dict:
    """Calcule les parametres optimaux d'un filtre de Bloom."""
    m = -n * math.log(p) / (math.log(2) ** 2)
    k = (m / n) * math.log(2)
    return {
        "n_elements": n,
        "p_faux_positifs": p,
        "m_bits": int(m),
        "m_octets": int(m / 8),
        "k_fonctions_hash": round(k),
    }

print(bloom_params(1_000_000, 0.01))
# {'n_elements': 1000000, 'p_faux_positifs': 0.01,
#  'm_bits': 9585059, 'm_octets': 1198132, 'k_fonctions_hash': 7}

print(bloom_params(10_000_000, 0.001))
# m_octets ~ 17 MB pour 10M elements avec p=0.1%

import hashlib
from bitarray import bitarray

class BloomFilter:
    def __init__(self, m: int, k: int):
        self.m = m
        self.k = k
        self.bits = bitarray(m)
        self.bits.setall(0)

    def _positions(self, item: str):
        for i in range(self.k):
            h = hashlib.sha256(f"{i}:{item}".encode()).hexdigest()
            yield int(h, 16) % self.m

    def add(self, item: str):
        for pos in self._positions(item):
            self.bits[pos] = 1

    def might_contain(self, item: str) -> bool:
        return all(self.bits[pos] for pos in self._positions(item))

bf = BloomFilter(m=9_585_059, k=7)
bf.add("user@example.com")
print(bf.might_contain("user@example.com"))   # True (certain)
print(bf.might_contain("other@example.com"))  # False ou True (faux positif possible)

Cas d'usage¶

Détection de doublons : avant d'inserer une URL dans un crawler, vérifier dans le Bloom filter si elle a déjà été visitee. Un faux positif signifie juste qu'on rate une URL — acceptable. Un faux negatif signifierait revisiter des URL — impossible avec un Bloom filter.

Réduction des accès disque : HBase, Cassandra et LevelDB maintiennent un Bloom filter par SSTable. Avant de lire une SSTable pour chercher une clé, on teste le Bloom filter. S'il répond "absent", on evite une lecture disque coûteuse.

Filtre de spam : tester si un expéditeur est dans une liste noire sans charger la liste complète en mémoire.

Scalable Bloom Filter et union¶

Les filtres de Bloom standards ont une capacité fixe — ajouter trop d'éléments augmente le taux de faux positifs. Le Scalable Bloom Filter chaîne plusieurs filtres avec des taux de faux positifs de plus en plus faibles, maintenant le taux global sous un seuil même quand la capacité explose.

L'union de deux filtres de Bloom de même taille et mêmes parametres se fait par OR bit-a-bit. L'intersection (AND bit-a-bit) est possible mais surestime la taille de l'intersection — utiliser avec précaution.

HyperLogLog : compter les éléments distincts¶

Compter les éléments distincts (cardinality estimation) dans un flux est un problème fondamental : combien d'utilisateurs uniques ont visite le site ? Combien de requêtes SQL distinctes ont été exécutées ? Une solution exacte par hachage nécessité \(O(n)\) mémoire. HyperLogLog le fait en \(O(1)\) — quelques kilooctets — avec une erreur standard de \(\approx 2\%\).

Principe intuitif¶

HyperLogLog exploite une propriété statistique des fonctions de hachage : si les hash sont uniformes, la probabilite qu'un hash commence par k zeros consecutifs est de \(1/2^k\). Si on observe un hash commencant par 10 zeros, il a fallu en moyenne \(2^{10} = 1024\) éléments distincts pour le voir.

L'algorithme divise les éléments en m sous-groupes (via les premiers bits du hash), maintient le nombre maximum de zeros initiaux observe dans chaque sous-groupe, et combine ces estimations via une moyenne harmonique.

Avec \(m = 2^{14} = 16384\) registres d'un octet chacun, on obtient :

• Mémoire : 16 KB
• Erreur standard : \(1.04 / \sqrt{m} \approx 0.81\%\)
• Plage : de 0 a \(2^{64}\) éléments distincts

HyperLogLog dans Redis¶

Redis implémenté HyperLogLog nativement avec les commandes PFADD, PFCOUNT, et PFMERGE.

# Ajouter des elements
PFADD visits:2024-01-15 "user:1" "user:2" "user:3" "user:1"
# (integer) 1

PFADD visits:2024-01-16 "user:2" "user:4" "user:5"
# (integer) 1

# Compter les distincts sur un jour
PFCOUNT visits:2024-01-15
# (integer) 3  (et non 4 — les doublons sont elimines)

# Merger plusieurs HLL — combien d'uniques sur 2 jours ?
PFMERGE visits:2024-01-15-16 visits:2024-01-15 visits:2024-01-16
PFCOUNT visits:2024-01-15-16
# (integer) 5  (union des distincts)

Probleme : compter les IPs uniques dans 100M de requetes

Solution exacte (set Python) :
  ~100M * 16 octets (UUID) = ~1.6 GB minimum

HyperLogLog (Redis) :
  12 KB fixe — quelle que soit la taille du jeu de donnees
  Erreur ~0.81%

Choix : si 0.81% d'erreur est acceptable (analytics, monitoring),
HyperLogLog est clairement superieur.
Si le compte exact est requis (facturation), utiliser un set.

CRDTs : Conflict-free Replicated Data Types¶

Dans un système distribué, plusieurs répliqués d'une même donnée peuvent être modifiées simultanément sans coordination. À la reconciliation, comment fusionner les modifications sans perdre d'information ? Les CRDTs (Conflict-free Replicated Data Types) sont des structures mathematiquement conccues pour que la fusion soit toujours commutative, associative et idempotente — garantissant la convergence quelle que soit l'ordre d'application des opérations.

G-Counter (Grow-only Counter)¶

Un compteur qui ne peut qu'augmenter. Chaque nœud maintient son propre compteur. La valeur globale est la somme de tous les compteurs. La fusion prend le maximum de chaque nœud.

from dataclasses import dataclass, field

@dataclass
class GCounter:
    node_id: str
    counts: dict = field(default_factory=dict)

    def increment(self, amount: int = 1):
        self.counts[self.node_id] = self.counts.get(self.node_id, 0) + amount

    def value(self) -> int:
        return sum(self.counts.values())

    def merge(self, other: "GCounter") -> "GCounter":
        merged = GCounter(self.node_id)
        all_nodes = set(self.counts) | set(other.counts)
        for node in all_nodes:
            merged.counts[node] = max(
                self.counts.get(node, 0),
                other.counts.get(node, 0)
            )
        return merged

# Deux noeuds incrementent independamment
a = GCounter("node-A")
b = GCounter("node-B")
a.increment(5)
b.increment(3)

merged = a.merge(b)
print(merged.value())  # 8 — sans coordination reseau

@dataclass
class PNCounter:
    """Positive-Negative Counter — peut incrementer et decrementer."""
    node_id: str
    pos: GCounter = None
    neg: GCounter = None

    def __post_init__(self):
        self.pos = GCounter(self.node_id)
        self.neg = GCounter(self.node_id)

    def increment(self, amount: int = 1):
        self.pos.increment(amount)

    def decrement(self, amount: int = 1):
        self.neg.increment(amount)

    def value(self) -> int:
        return self.pos.value() - self.neg.value()

    def merge(self, other: "PNCounter") -> "PNCounter":
        result = PNCounter(self.node_id)
        result.pos = self.pos.merge(other.pos)
        result.neg = self.neg.merge(other.neg)
        return result

LWW-Register (Last-Write-Wins Register)¶

Un registre ou la dernière écriture (par timestamp) gagne. Simple mais sujet aux conflits si les horloges ne sont pas synchronisées. Utilisé dans Cassandra (avec timestamp applicatif) et Amazon DynamoDB (avec version vectors).

from dataclasses import dataclass
from typing import Any
import time

@dataclass
class LWWRegister:
    value: Any = None
    timestamp: float = 0.0

    def write(self, value: Any, ts: float = None):
        ts = ts or time.time()
        if ts > self.timestamp:
            self.value = value
            self.timestamp = ts

    def merge(self, other: "LWWRegister") -> "LWWRegister":
        result = LWWRegister()
        if self.timestamp >= other.timestamp:
            result.value = self.value
            result.timestamp = self.timestamp
        else:
            result.value = other.value
            result.timestamp = other.timestamp
        return result

OR-Set (Observed-Remove Set)¶

Un ensemble ou les ajouts et suppressions peuvent coexister sans conflits. Chaque élément est taggue avec un identifiant unique à l'ajout. La suppression n'efface que les tags observes, pas les ajouts futurs — resolvant le problème "add gagne sur remove" ou "remove gagne sur add" de manière deterministe.

from dataclasses import dataclass, field
import uuid

@dataclass
class ORSet:
    """Observed-Remove Set — les ajouts concurrents aux suppressions survivent."""
    adds: dict = field(default_factory=dict)     # element -> set(tags)
    removes: set = field(default_factory=set)    # tags supprimes

    def add(self, element):
        tag = str(uuid.uuid4())
        if element not in self.adds:
            self.adds[element] = set()
        self.adds[element].add(tag)
        return tag

    def remove(self, element):
        if element in self.adds:
            self.removes.update(self.adds[element])

    def contains(self, element) -> bool:
        tags = self.adds.get(element, set())
        return bool(tags - self.removes)

    def merge(self, other: "ORSet") -> "ORSet":
        result = ORSet()
        # Union des ajouts
        all_elements = set(self.adds) | set(other.adds)
        for e in all_elements:
            result.adds[e] = self.adds.get(e, set()) | other.adds.get(e, set())
        # Union des suppressions
        result.removes = self.removes | other.removes
        return result

Applications en production¶

Amazon DynamoDB : hachage consistant en détail¶

DynamoDB partition les données sur des "storage nodes" via un anneau de hachage consistant. La clé primaire d'un item est hachee (Murmur3) pour déterminer son storage node. Chaque item est répliqué sur 3 nœuds pour la durabilité. Quand un nœud est ajoute (scaling horizontal), seules les clés de la portion de l'anneau précédemment gérée par le voisin migrent.

Cassandra : vnodes et réplication¶

Cassandra utilisé par défaut 256 vnodes par nœud physique. Le token ring est de taille \(2^{64}\). La "réplication strategy" copie chaque clé sur les N nœuds successeurs dans l'anneau (N = réplication factor). Le NetworkTopologyStrategy assure que les répliqués sont réparties sur plusieurs datacenters — une panne de datacenter ne perd pas de données.

Recap¶