cache, persist et checkpoint en Spark : quoi garder ?

Spark ne garde rien entre deux actions. Chaque count(), chaque write, chaque collect() rejoue le plan complet depuis la source — c’est la conséquence directe de l’évaluation paresseuse. Sur un pipeline qui lit 500 Go et fait trois actions, vous lisez 1,5 To.

La réponse s’appelle cache() ou persist(). Mais mal utilisée, elle coûte plus qu’elle ne rapporte : de la mémoire prise à l’exécution, des partitions évincées, des jobs qui ralentissent au lieu d’accélérer. Voici quand cacher, à quel niveau, et surtout quand ne pas le faire.

Le problème que ça résout

val errors = logs.filter($"level" === "ERROR")

println(errors.count())        // job 1 : lit + filtre 500 Go
errors.write.parquet("/out")   // job 2 : relit + refiltre 500 Go

Deux actions, deux lectures complètes. Avec un cache :

val errors = logs.filter($"level" === "ERROR").cache()

println(errors.count())        // matérialise et garde
errors.write.parquet("/out")   // réutilise ce qui est en mémoire

Le gain est réel dès deux réutilisations. En dessous, cache() ajoute un coût sans bénéfice.

Le piège : cache() ne fait pas la même chose sur RDD et DataFrame

C’est l’erreur la plus répandue, y compris dans la documentation officieuse :

Conséquence pratique : sur un RDD, si la mémoire manque, les partitions qui ne tiennent pas sont simplement recalculées à chaque accès — silencieusement, sans erreur, avec un job qui devient mystérieusement lent. Sur un DataFrame, elles débordent sur disque, ce qui est plus lent que la mémoire mais bien plus rapide qu’un recalcul complet.

Sur un RDD dont vous n’êtes pas certain qu’il tienne en mémoire, soyez explicite :

import org.apache.spark.storage.StorageLevel

rdd.persist(StorageLevel.MEMORY_AND_DISK)

Les niveaux de stockage

import org.apache.spark.storage.StorageLevel
df.persist(StorageLevel.MEMORY_AND_DISK_SER)

La sérialisation (_SER) divise typiquement l’empreinte mémoire par deux à cinq, au prix de cycles CPU à chaque lecture. Sur un cluster où la mémoire est la ressource rare — le cas courant — c’est presque toujours un bon échange.

À noter pour les DataFrames : le format interne Tungsten est déjà compact et binaire, donc l’écart entre les variantes sérialisées et non sérialisées y est bien plus faible que sur les RDD.

Quand cacher, vraiment

Trois situations le justifient, et seulement trois :

1. Le DataFrame est utilisé par au moins deux actions. C’est le critère principal. Une seule action ⇒ aucun cache.
2. La lignée en amont est coûteuse — plusieurs shuffles, un join large, une lecture de milliers de fichiers. Recalculer coûte cher.
3. Algorithme itératif — descente de gradient, PageRank, tout ce qui reboucle sur le même dataset. C’est le cas d’usage historique de Spark.

Et l’anti-pattern symétrique :

// Inutile : une seule action derrière
val df = spark.read.parquet(path).filter(...).cache()
df.write.parquet(out)

Ici le cache ajoute une écriture mémoire complète pour rien. Le job est plus lent qu’avec la version sans cache.

unpersist() : la moitié que tout le monde oublie

Un cache occupe la mémoire de l’exécuteur jusqu’à la fin de l’application, ou jusqu’à ce que Spark l’évince pour faire de la place. Dans un pipeline long, les caches accumulés étranglent progressivement les étapes suivantes.

val intermediaire = grosCalcul.cache()

etape1(intermediaire)
etape2(intermediaire)

intermediaire.unpersist()   // libère explicitement

unpersist() est bloquant par défaut sur les DataFrames et non bloquant sur les RDD ; unpersist(blocking = true) garantit que la mémoire est rendue avant de continuer.

Vérifiez ce qui occupe la mémoire dans l’onglet Storage de l’UI Spark : chaque dataset caché y apparaît avec sa taille, son niveau et son pourcentage effectivement en mémoire. Un dataset affiché à « 60 % cached » signifie que 40 % sera recalculé à chaque accès — signal clair qu’il faut passer en MEMORY_AND_DISK ou réduire le volume caché.

val df = source.filter(...).cache()
// Rien en mémoire à cet instant.

df.count()   // ← C'est ici que le cache se remplit.
df.count()   // ← Celui-ci lit le cache.

checkpoint() : quand cache() ne suffit pas

cache() garde les données mais conserve toute la lignée — la recette qui permet de recalculer en cas de perte d’un exécuteur. Sur un pipeline itératif de cent tours de boucle, cette lignée grossit à chaque itération, jusqu’à ce que le plan devienne ingérable : plan de milliers de nœuds, StackOverflowError à l’analyse, driver qui passe plus de temps à optimiser qu’à exécuter.

checkpoint() tronque la lignée en écrivant physiquement sur un stockage fiable :

sc.setCheckpointDir("hdfs:///tmp/checkpoints")   // ou s3a://, obligatoire

val df = calculIteratif(...)
df.checkpoint()   // écrit sur disque et oublie la lignée

Combinaison recommandée quand vous checkpointez : cacher avant, pour éviter que le checkpoint ne déclenche un recalcul complet du plan.

df.cache()
df.checkpoint()

Ce qu’il faut retenir

cache() répond à une question précise : « ce dataset va-t-il servir au moins deux fois ? » Si la réponse est non, il coûte sans rapporter. Si elle est oui, MEMORY_AND_DISK est le niveau par défaut raisonnable — jamais MEMORY_ONLY sur un RDD dont vous ignorez la taille, sous peine de recalculs silencieux.

Deux réflexes qui distinguent un pipeline soigné : unpersist() dès que le dataset n’est plus utile, et un coup d’œil à l’onglet Storage de l’UI pour vérifier que ce qui est censé être en mémoire y est réellement. Et checkpoint() uniquement pour ce qu’il fait de spécifique : couper une lignée devenue trop longue, ce qu’aucun cache ne sait faire.