Modèles de calculs parallèles distribués pour Python scientifique : MPI, DASK et RAY

Vous disposez d’une charge de travail de calcul qui ne tient pas sur un seul cœur. Il s’agit peut-être d’une simulation PDE à grande échelle, d’un lot de balayages de paramètres ou d’un pipeline de traitement de données qui prend des heures au lieu de minutes. Vous avez entendu parler de MPI, Dask et Ray comme outils de pointe pour mettre à l’échelle le code Python. Mais lequel correspond réellement à votre problème ?

La réponse courte : cela dépend de votre modèle de charge de travail. MPI vous donne le plus de contrôle mais a la courbe d’apprentissage la plus raide. DASK vous aide à faire évoluer les pandas familiers et le code NumPy avec des modifications minimales. Ray fonctionne mieux pour les charges de travail mixtes où la formation ML, la planification dynamique et l’état distribué existent dans le même flux de travail.

La plupart des chercheurs commencent par Dask car il correspond directement au code qu’ils connaissent déjà. Ils passent à Ray lorsqu’ils ont besoin d’une formation de modèle ou d’une orchestration dynamique. Ils choisissent MPI lorsqu’ils travaillent avec la décomposition de domaine pour les PDE couplées et ont besoin d’un contrôle de performance de bas niveau.

Ce guide explique les modèles de calcul parallèles derrière chaque cadre. Il vous aide à choisir le bon outil tôt au lieu de reconstruire votre flux de travail après que la mauvaise implémentation est devenue trop difficile à mettre à l’échelle.

Ce qui rend ces cadres différents

Les trois outils résolvent le même problème de base : la distribution du calcul Python sur plusieurs cœurs ou machines. Ils utilisent des approches très différentes parce qu’elles ont été conçues pour différents écosystèmes. Ces différences de conception décident si votre code s’exécute en quelques minutes ou devient difficile à maintenir.

MPI, ou interface de passage de messages, a été conçu pour le calcul haute performance et le calcul scientifique. Il suit un modèle de transmission de messages où les processus s’exécutent de manière indépendante et communiquent explicitement. Les développeurs Python l’utilisent généralement via mpi4py, qui enveloppe la bibliothèque C MPI. Ce modèle est explicite, de bas niveau et donne un contrôle précis sur la façon dont les données se déplacent entre les processus.

Dask a été construit pour l’écosystème de la science des données Python. Il fournit des versions parallèles des tableaux NumPy, des trames de données Pandas et des estimateurs de Scikit-Learn. Il utilise une évaluation paresseuse. Votre code construit un graphe de tâches et Dask optimise et exécute ce graphe en parallèle. Si vous écrivez déjà des pandas ou du code numpy, Dask ne nécessite souvent qu’une petite modification d’importation.

Ray a été conçu pour les applications Python évolutives avec des charges de travail mixtes. Il utilise deux primitives principales : les tâches pour l’exécution de fonctions sans état et les acteurs pour les objets distribués avec état. Ray utilise également un magasin d’objets à mémoire partagée pour déplacer efficacement les données entre les nœuds. Il a été construit avec des flux de travail d’apprentissage automatique à l’esprit, de sorte qu’il comprend des outils pour la formation, le réglage et les modèles de service.

Quand utiliser MPI pour le code scientifique Python

MPI est un outil standard en sciences informatiques. Si votre recherche implique la dynamique des fluides, la mécanique solide, les simulations en champ de phase ou les méthodes numériques qui décomposent un domaine entre les processeurs, MPI vous offre un outillage mature et un soutien communautaire solide.

Le modèle SPMD

MPI suit le modèle de données à programme unique et multiple. Chaque processus exécute le même code mais fonctionne sur des données différentes. La communication se fait par le biais d’opérations explicites point à point ou collectives.

Avec mpi4py, une simulation parallèle de base peut ressembler à ceci :

from mpi4py import MPI
import numpy as np

comm = MPI.COM_WORLD
rank = comm.Get_rank()
size = comm.Get_size()

# Each process handles a different slice of the domain
local_data = create_domain_slice(rank, size)
result = solve_pde(local_data)

# Collective communication to gather results
all_results = comm.gather(result, root=0)

if rank == 0:
    # Assemble global solution
    global_solution = assemble(all_results)
    report_results(global_solution)

Quand MPI a du sens

Utilisez MPI pour la décomposition du domaine dans les simulations couplées. Si les processus doivent communiquer souvent au niveau numérique, MPI donne le contrôle requis pour gérer ces modèles de communication. Ceci est courant dans les EDP couplés, l’échange de conditions limites et les problèmes d’interface partagée.

MPI fonctionne également bien pour le couplage numérique dense. Si votre solveur a besoin d’une coordination étroite de processus, comme une itération Newton avec des résidus globaux, des opérations collectives MPI comme Allreduce, Bcast et Scatter sont conçues pour ce type de travail.

MPI est également le bon choix lorsque les performances maximales sont importantes. Il s’exécute directement sur l’infrastructure HPC sans couche d’abstraction de haut niveau entre votre code et le matériel. Cela donne de bonnes performances, mais cela signifie également que vous gérez vous-même le parallélisme, la communication et l’équilibre de la charge.

Le compromis

Le code MPI peut devenir verbeux. Chaque opération nécessite des appels explicites pour envoyer, recevoir, diffuser ou collecter des données. Il n’y a pas d’optimisation automatique des graphes de tâches. Vous concevez vous-même la structure de communication.

Cette complexité est acceptable lorsque la performance est la principale priorité. Il est moins attrayant lorsque vous n’avez besoin de tester que si la parallélisation favorise un flux de travail de recherche. Une règle pratique est simple : choisissez MPI lorsque vous écrivez une bibliothèque de solveur ou un code de simulation de production où les performances justifient le coût de mise en œuvre.

Quand utiliser Dask pour le code scientifique Python

Dask a été conçu pour mettre à l’échelle la pile de données Python existante sans vous obliger à tout réécrire. Si vous utilisez déjà Pandas, NumPy ou Scikit-Learn, Dask vous permet souvent de conserver le même modèle mental tout en ajoutant une exécution parallèle.

Le modèle d’évaluation paresseux

Dask construit des graphes de tâches paresseusement. Lorsque vous appelez des opérations DASK, le calcul ne s’exécute pas immédiatement. Au lieu de cela, vous décrivez ce qui devrait arriver. Dask optimise ensuite le graphique, planifie le travail entre les travailleurs et l’exécute en parallèle.

Ce modèle paresseux présente deux avantages importants :

1. Optimisation du graphique des tâches. Dask peut combiner des opérations, supprimer le calcul redondant et réorganiser les tâches pour de meilleures performances.
2. Gestion de la mémoire. Le calcul étant retardé, Dask peut gérer plus efficacement les résultats intermédiaires.

Voici le modèle de base :

import dask.dataframe as dd

# Lazy: no computation happens yet
df = dd.read_parquet("simulations/*.parquet")
filtered = df[df.temperature > 300]
aggregated = filtered.groupby("region").mean()

# This triggers computation across the cluster
results = aggregated.compute()

Quand Dask a du sens

DASK fonctionne bien pour les transformations de tableaux et de trames de données à grande échelle. Si votre workflow implique une agrégation de données de simulation, une ingénierie des fonctionnalités ou une analyse par lots sur des données structurées, DASK s’associe naturellement aux pandas existants et aux flux de travail NumPy.

Dask est également fort lorsque le flux de travail a un graphique de tâches prévisible. Un modèle commun est : lire des données, transformer des données, agréger les résultats et écrire la sortie. DASK peut optimiser efficacement cette structure.

Il est également utile pour la parallélisation progressive. Vous pouvez commencer par un flux de travail Pandas mono-machine et passer ultérieurement à une exécution distribuée. Cela fait de Dask un point de départ pratique pour les chercheurs qui souhaitent un traitement parallèle sans réécriture complète.

Le compromis

DASK est moins adapté aux charges de travail hautement dynamiques ou mixtes. Si votre pipeline comprend une formation de modèles, une planification dynamique ou des services à long terme, Dask n’est peut-être pas le mieux adapté. Son modèle de graphe de tâches fonctionne mieux lorsque la structure du flux de travail est connue à l’avance.

DASK n’est pas non plus à l’échelle aussi efficace que le MPI pour un couplage numérique serré. Si votre simulation nécessite un échange fréquent de conditions limites entre les processus, MPI donne généralement de meilleures performances de bas niveau.

Une remarque pratique : la communication DASK peut être plus lente sur les réseaux à haute latence. Si vous utilisez un cluster avec une interconnexion à faible latence, MPI peut être plus performant. Sur les systèmes cloud ou les réseaux Ethernet standard, Dask est souvent suffisant pour de nombreux flux de travail de recherche.

Quand utiliser Ray pour le code scientifique Python

Ray utilise un modèle différent. Au lieu de se concentrer principalement sur les graphes de tâches, il utilise des tâches et des acteurs. Les tâches exécutent des fonctions parallèles sans état. Les acteurs sont des objets distribués qui maintiennent l’état entre les appels de méthode.

Le modèle d’acteur

Les acteurs sont l’une des caractéristiques les plus importantes de Ray. Un acteur vit sur un nœud du cluster et conserve son état interne entre les appels. Ceci est utile lorsque différentes parties d’un flux de travail distribué ont besoin d’un état persistant.

import ray

ray.init()

@ray.remote
class SimulationState:
    def __init__(self):
        self.state = initialize_state()
    
    def step(self, local_data):
        self.state = evolve(self.state, local_data)
        return self.state
    
    def get_snapshot(self):
        return self.state

# Actor runs on a specific node
sim = SimulationState.remote()
result = sim.step.remote(local_data)
snapshot = ray.get(sim.get_snapshot.remote())

Quand Ray a du sens

Ray fonctionne bien pour les charges de travail hétérogènes. Si votre pipeline de simulation comprend le traitement des données, la formation de modèles, le réglage des hyperparamètres et la diffusion de modèles, Ray peut coordonner ces parties dans un seul système.

Ray est également utile pour la planification dynamique. Si la structure de votre calcul change lors de l’exécution, Ray peut s’adapter. Cela aide dans les flux de travail tels que le raffinement du maillage adaptatif, où de nouvelles tâches peuvent apparaître en fonction des résultats intermédiaires.

Ray prend également en charge des charges de travail mixtes CPU et GPU. Si vous exécutez des solveurs basés sur CPU avec des modèles de post-traitement ou de substitution accélérés par GPU, Ray peut planifier des travaux sur différentes ressources matérielles.

Un autre cas d’utilisation fort est l’orchestration des expériences. Ray peut gérer de nombreuses configurations de simulation, suivre l’état des exécutions et coordonner les résultats distribués.

Le compromis

Ray a besoin d’une gestion prudente du cycle de vie. Les acteurs doivent être créés, utilisés et libérés correctement. Si les acteurs conservent de grandes structures de données pendant trop longtemps, l’utilisation de la mémoire peut s’allonger dans le cluster.

Ray est également moins idéal pour les transformations de données déterministes pures par lots et simples. Dans ces cas, Dask semble souvent plus naturel et peut nécessiter moins de code.

Un point de décision utile est le suivant : Ray est attrayant lorsque la vitesse locale n’est pas la seule préoccupation. Sa valeur réelle apparaît lorsque les charges de travail deviennent distribuées, dynamiques, avec état ou apprentissage automatique.

Comment choisir entre MPI, Dask et Ray

Vous n’avez pas toujours besoin de choisir un seul cadre. De nombreuses équipes de recherche utilisent une approche hybride. Chaque framework gère la partie du flux de travail qu’il s’adapte le mieux.

flux de décision

Étape 1 : Votre charge de travail est-elle principalement des transformations de tableaux ou de dataframes ?

• Considérez DASK. Il convient parfaitement à l’agrégation de données de simulation, à l’ingénierie des fonctionnalités, à l’analyse et aux pipelines de données structurées.

Étape 2 : Votre charge de travail inclut-elle la planification dynamique, la formation de modèles ou les composants avec état ?

• Considérez Ray. Il s’adapte aux workflows qui ont besoin d’acteurs, d’un état distribué, de GPU ou de structures de tâches.

Étape 3 : Avez-vous besoin d’un couplage numérique ou d’une décomposition de domaine serrés pour les PDE ?

• Considérez MPI. Il est conçu pour une communication de processus fréquente et un contrôle de bas niveau.

Étape 4 : Avez-vous plusieurs types de charge de travail ?

• Envisagez une architecture hybride. Vous pouvez utiliser Dask pour la préparation des données, Ray pour l’orchestration et MPI pour le solveur numérique.

Recommandations pratiques

Le cas des architectures hybrides

Vous n’avez pas besoin d’utiliser un framework pour l’ensemble du workflow. Certaines architectures de calcul scientifique solides combinent plusieurs outils :

• MPI pour le solveur, dask pour le post-traitement. Exécutez le solveur PDE avec MPI, puis analysez et visualisez les résultats avec Dask.
• MPI pour le solveur, Ray pour l’orchestration d’expérience. Utilisez MPI pour la simulation numérique et le rayon pour gérer les configurations, l’état et plusieurs exécutions.
• DASK pour la préparation des données, Ray pour la formation des modèles. Utilisez DASK pour préparer de grands ensembles de données, puis utilisez Ray pour la formation de modèles ou le développement de modèles de substitution.

erreurs courantes

Une erreur courante consiste à supposer que Dask met à l’échelle chaque type de charge de travail. DASK est solide pour les flux de travail déterministes avec des graphes de tâches clairs. Si la structure de la tâche dépend de résultats intermédiaires, Ray peut mieux gérer le flux de travail.

Une autre erreur consiste à utiliser MPI où Dask serait plus simple. Si la charge de travail est principalement constituée d’analyse de données ou d’ingénierie de fonctionnalités, Dask peut gagner beaucoup de temps de mise en œuvre.

Certaines équipes négligent également la tolérance aux pannes. MPI donne un contrôle manuel, mais vous devez concevoir vous-même la gestion des échecs. DASK et RAY offrent une prise en charge plus importante au niveau du planificateur pour les tentatives et la récupération.

Les frais généraux de communication sont un autre problème. Chaque cadre distribué a des coûts de réseau. MPI rend la communication visible et plus facile à optimiser. Dask et Ray cachent une grande partie de cette complexité, mais le coût existe toujours.

Ce que nous recommandons

Pour la plupart des chercheurs scientifiques en Python, commençant par l’informatique distribuée, Dask est le meilleur point d’entrée. Il mappe sur le code qu’ils écrivent déjà, nécessite moins de changements et donne un parallélisme distribué sans un modèle de programmation complètement nouveau.

Choisissez MPI lorsque vous écrivez du code de simulation de production où les performances justifient le coût de mise en œuvre. C’est le bon outil pour la décomposition du domaine et le couplage numérique serré.

Choisissez Ray lorsque le flux de travail est hétérogène. Ray est fort lorsque le traitement des données, la formation de modèles, la planification dynamique et l’état distribué apparaissent dans le même système.

Les cadres ne s’excluent pas mutuellement. De nombreuses équipes utilisent chaque framework pour la partie du flux de travail qu’il gère le mieux. La clé est de faire correspondre le modèle de charge de travail à l’abstraction droite.

Résumé

Le choix entre MPI, Dask et Ray n’est pas de savoir quel framework est le meilleur en général. Il s’agit de la structure qui correspond à votre charge de travail.

• MPI donne un contrôle de bas niveau pour un couplage numérique serré et une décomposition de domaine.
• Dask met à l’échelle des pandas familiers et des flux de travail NumPy avec des graphes de tâches paresseux.
• Ray gère des charges de travail hétérogènes avec des tâches, des acteurs, une planification dynamique et un état distribué.

Commencez par Dask si votre charge de travail s’apparente à des transformations Array ou DataFrame. Utilisez MPI si vous avez besoin de décomposition de domaine ou de couplage numérique serré. Choisissez Ray si votre pipeline inclut une formation de modèle, une planification dynamique ou un état distribué.