Traiter de grands volumes de données Doctrine grâce à Messenger
Jean-Pascal, Architecte web 9 avril 2020
Jean-Pascal, Architecte web 9 avril 2020
Comme promis dans mon précédent article "Rex : RabbitMQ et Supervisor avec Docker et le composant Messenger de Symfony", je reviens vous parler de grands volumes de données avec Doctrine.
Et à priori je ne peux pas t’en vouloir, c’est même marqué dans la doc que c’est une mauvaise idée.
Sauf que moi, j’aime quand même bien Doctrine, surtout quand je dois créer une hiérarchie un peu complexe, avec des comportements du style de Gedmo. Bref, le genre de trucs pas fun à simuler en SQL pur.
Deux des gros problèmes des longs scripts Doctrine, qui expliquent leur lenteur, sont :
Pour limiter le temps de chaque script, la solution qui m’est naturellement apparue était la pagination. Si on on considère que chaque page consiste à créer X objets aléatoires, chaque script ne s’occuperait que d’une partie bien délimitée de la masse de données à traiter, et le script suivant d’une autre partie.
Ainsi l’EntityManager est rechargé à chaque page, et ne tourne pas trop longtemps.
De plus, en utilisant les propriétés de la congruence, il est possible de paralléliser le traitement de plusieurs pages (tu vas voir, c’est plus simple que ce jargon mathématique ne le laisse penser) :
Pour paralléliser n pages il suffit de :
On a donc récursivement le traitement parallèle de toutes les pages de l’ensemble.
C’est la que Messenger entre en scène.
En effet, en partant du principe que chaque message contient les informations nécessaires au filtrage (l’entité concernée, etc.), la page concernée (le i) ainsi que le nombre de traitement simultanés (le n), on peut considérer le traitement de chaque page comme le simple traitement d’un message, et à la fin du traitement, renvoyer le même message pour la page i+n.
En combinant cette logique avec Supervisor pour pouvoir limiter le temps de vie de chaque consumer, on peut simplement traiter des scripts de plusieurs heures.
Et on peut utiliser cette simple logique pour également traiter de gros imports, dans le cadre d’une reprise d’existant par exemple.
C’est souvent le cas si quand on fait des imports / des fixtures : on a besoin de faire un autre traitement, mais seulement une fois que le premier script est terminé avec succès. Du fait de la parallélisation des traitements, il devient complexe de savoir si cette condition est remplie.
Complexe, certes. Impossible, sûrement pas.
L’approche que j’ai adoptée repose sur plusieurs prérequis :
public function __invoke(Message $message): void
{
try {
// fait les traitements nécessaires, et retourne vrai si on considère que cette page est une des pages de fin
$over = $this->handleExecution($message,$debut, $fin);
if (!$over) {
// On a pas terminé la série en cours, on va donc traiter la page n + step
$message->setPage($message->getPage() + $message->getStep());
$this->bus->dispatch($message);
return;
}
$importKey = $message->getImportKey();
// On utilise la date de fin, qui est toujours définie, et la même pour tous les messages concurrents dans tous les cas
$count = $this->updateCount($importKey, $fin);
if ($count < $message->getPageStep()) {
// On a pas encore terminé l'import, il reste des séries
return;
}
// toutes les séries de l'import sont arrivées à complétion, et on doit lancer le prochain import
$key = \array_search($importKey, $this->importSequence, true);
if ($key === false) {
// on a pas trouvé la clef, on ne pourra pas trouver de suivant, donc on sort
return;
}
if (\array_key_exists($key + 1, $this->importSequence)) {
$nextKey = $this->importSequence[$key + 1];
// on a trouvé la clef, on lance donc les séries de l'import suivant
for ($startingPage = 1; $startingPage < 1 + $message->getPageStep(); $startingPage++) {
$this->bus->dispatch(new Message(
$nextKey,
$fin,
null,
$startingPage,
$message->getPageStep()
));
}
} else {
// il n'y a pas de clef suivante : on est donc à la fin de tous les imports
// on peut éventuellement faire des traitements de fin
}
}
Et la partie sur le cache et le lock que j’ai teasée juste avant me diras-tu ? Et bien, c’est utilisé pour la gestion du compteur de pages terminées, dans le méthode updateCount :
public function updateCount($importKey, $fin): int
{
$dateAsString = $fin->format("U.u");
// on doit mettre un lock ici, pour s'assurer que deux threads concurrents d'import ne vont pas incrémenter le count n'importe comment
// @var Symfony\Component\Lock\LockInterface $lock
$lock = $this->lockFactory->createLock($importKey);
$lock->acquire(true);
// on recupère la valeur stockée, avec null par défaut
$item = $this->cache->get($importKey, static function () {
return null;
});
// puis on la supprime
// @var Symfony\Contracts\Cache\CacheInterface $this->cache
$this->cache->delete($importKey);
if (!isset($item[$dateAsString])) {
// la date a changé, on est donc sur un autre import, il faut repartir a 1
$item = $this->cache->get($importKey, static function () use ($dateAsString) {
return [$dateAsString => 1];
});
} else {
// la date est la même, on incrémente donc de 1
$item = $this->cache->get($importKey, static function () use ($dateAsString, $item) {
return [$dateAsString => $item[$dateAsString] + 1];
});
}
// on relache le lock pour pouvoir
$lock->release();
return $item[$dateAsString];
}
On peut voir dans cette méthode l’utilisation du Cache Component et du Lock Component sans avoir besoin d’expliciter l’implémentation du cache (en l’occurence j’aime bien l’implémentation doctrine dans la mesure où elle permet de vérifier les valeurs utilisées, mais ça fonctionne exactement de la même façon avec n’importe quelle implémentation).
Grâce à ces quelques astuces, on a vu que malgré les mises en garde, il était finalement possible de traiter de grands volumes de données en tirant partie de la parallélisation.
