Combicancer : pipeline de pré-processing et de reformatage en format OMOP-like - OMOPization du SNDS natif.

Approches et Raisonnement

• Traitement Distribué avec PySpark : Le choix de PySpark est central. Le volume des données du SNDS (plusieurs téraoctets) rend leur traitement sur une seule machine impossible. PySpark permet de distribuer les calculs sur un cluster, assurant la scalabilité et des performances adéquates pour les opérations de nettoyage, de jointure et de transformation.
• Format de Stockage Optimisé (Parquet) : Les données brutes, souvent en format CSV, sont systématiquement converties en Parquet. Ce format colonnaire est optimisé pour les requêtes analytiques. Il offre une meilleure compression et des vitesses de lecture significativement plus rapides que les formats basés sur des lignes comme le CSV, ce qui est crucial pour les étapes successives du pipeline qui lisent et manipulent de grands volumes de données.
• Modélisation des Relations par un Graphe : Une approche sophistiquée a été adoptée pour gérer les relations complexes entre les nombreuses tables du SNDS. Un graphe de relations (graph.pkl) est généré pour modéliser le schéma de la base de données. Cette structure permet d'automatiser et de simplifier le processus de "flattening" (aplatissement), en déterminant dynamiquement les chemins de jointure les plus courts entre une table donnée et la table de référence des patients (IR_BEN_R_uniq).
• Pipeline Modulaire et Étape par Étape : L'algorithme est structuré en modules distincts et en étapes séquentielles (nettoyage, création du graphe, aplatissement, post-traitement, création des tables finales). Cette modularité rend le code plus lisible, maintenable et testable. L'utilisation de dossiers intermédiaires (step1, step2, etc.) permet de sauvegarder les résultats de chaque grande étape, facilitant le débogage et la reprise du pipeline sans avoir à tout ré-exécuter.
• Qualité des Données : Une attention particulière est portée à la qualité des données. De nombreux scripts et notebooks sont dédiés à la validation et au nettoyage des données (snds_utils_spark.py, notebooks Q*.ipynb). Des filtres spécifiques sont appliqués pour corriger des problèmes connus dans les données du SNDS, tels que les doublons, les valeurs aberrantes ou les codes d'erreur.
   

Défis Relevés

• Volumétrie et Complexité : Le principal défi était de concevoir un système capable de traiter efficacement l'énorme volume de données du SNDS et sa structure relationnelle complexe. L'utilisation de PySpark et l'approche par graphe ont été des solutions clés.
• Hétérogénéité des Données : Les tables du SNDS (DCIR, PMSI) ont des structures et des conventions de nommage différentes. De plus, les schémas peuvent évoluer d'une année à l'autre. L'algorithme a dû être conçu pour gérer cette hétérogénéité, notamment en sélectionnant dynamiquement les colonnes disponibles et en appliquant des logiques de transformation adaptées à chaque source (par exemple, pour le calcul des dates de procédure dans les différentes tables du PMSI).
• Nettoyage des Données Spécifiques au SNDS : Le pipeline intègre la connaissance métier sur les particularités des données du SNDS. Par exemple, il gère la "régularisation" des quantités (prise en compte des factures rectificatives) et le nettoyage des identifiants patients (suppression des "jumeaux de même sexe").
• Optimisation des Performances : Pour éviter des calculs redondants, une stratégie d'optimisation a été mise en place lors de la phase d'aplatissement. Les jointures intermédiaires qui sont utilisées plusieurs fois sont sauvegardées temporairement pour accélérer le processus global.

V1 : octobre 2024

Prérequis

Avant de commencer, vous devez disposer des éléments suivants :

• Un environnement Python (version 3.8 ou supérieure est recommandée).
• Un cluster Apache Spark accessible. La configuration du projet (spark_config.json) suggère une utilisation avec Hive pour la gestion des métadonnées.
• Un compte de stockage Azure Blob Storage avec les informations d'identification nécessaires (ACCOUNT_NAME dans conf.py).
• Les données brutes du SNDS (fichiers CSV) disponibles dans votre compte de stockage Azure.
• Git pour cloner le projet depuis un dépôt.
   

Étapes d'installation
 

1. Cloner le projet

Utilisez Git pour cloner le projet sur votre environnement de travail, comme suggéré dans le fichier README.md.

Bash

cd existing_repo

git remote add origin https://gitlab.com/healthdatahub/boas/institutcurie/combicancer

git branch -M main

git push -uf origin main
 

2. Installer les dépendances Python

L'algorithme utilise plusieurs bibliothèques Python qui doivent être installées. Vous pouvez les installer via pip. Les principales dépendances, identifiées dans les notebooks 1-main_snds_processor_CLEANED.ipynb et 2-main_combicancer_pipeline_CLEANED.ipynb, sont :

• pyspark
• pandas
• adlfs (pour l'accès à Azure Data Lake Storage)
• azure-identity
• networkx
• pyyaml

Vous pouvez les installer avec la commande suivante :

Bash

pip install pyspark pandas adlfs azure-identity networkx pyyaml
 

3. Configurer l'environnement

Vous devez ensuite configurer le projet pour qu'il pointe vers votre environnement Spark et votre compte de stockage Azure.

1. Configuration de Spark : Modifiez le fichier spark_config.json pour y insérer les informations de connexion à votre cluster Spark.
   • "master": L'URL de votre master Spark (par exemple, spark://<master-ip>:<port>).
   • "metastore_uris": L'URI de votre metastore Hive.
   • "warehouse_dir": Le chemin vers votre "warehouse" Hive sur HDFS ou votre stockage cloud.
2. Configuration du Projet : Modifiez le fichier conf.py pour configurer les chemins de stockage.
   • ACCOUNT_NAME: Le nom de votre compte de stockage Azure.
   • PIPELINE_FOLDERS: Assurez-vous que les noms des dossiers intermédiaires (SRC_DATA, STEP1, etc.) correspondent à la structure de votre conteneur de stockage.
      

4. Préparer les données

Avant d'exécuter l'algorithme, vous devez :

1. Téléverser les données du SNDS dans le dossier source de votre conteneur de stockage Azure (défini par la variable SRC_LIVRAISON dans les notebooks principaux).

Créer les dossiers de destination : Le pipeline créera une série de dossiers pour stocker les résultats intermédiaires et finaux. Assurez-vous que les permissions d'écriture sont correctes.

Une fois l'environnement configuré et les données en place, vous pouvez exécuter les deux pipelines séquentiellement en utilisant les notebooks Jupyter fournis :

1. 1-main_snds_processor_CLEANED.ipynb : Ce notebook exécute la première grande étape de prétraitement des données brutes du SNDS.
2. 2-main_combicancer_pipeline_CLEANED.ipynb : Ce notebook prend en entrée les données prétraitées par le premier pipeline et génère les tables d'analyse finales.

Vous pouvez exécuter ces notebooks depuis un environnement Jupyter connecté à votre cluster Spark.

hdh@health-data-hub.fr

Maeva Kos (Plateforme des données de santé - Health Data Hub)
Paul Gougis (Institut Curie)