Spark Opitmisation

ENRICHISSEMENT DE DONNEES.
PREPARATION DES DONNEES POUR LA RECHERCHE DANS ELASTICSEARCH

J’ai une application web en scala pour faire du traitement de données en batch
via un pipeline ETL.
Le pipeline est composé de :
– Spark pour l’ingestion et le traitement en batch des données.
– MinIO un moteur de stockage objet. C’est la version open-source de S3
installable on-premises
– Elastic Search pour l’indexation des données et leur préparation pour une
recherche rapide
Les données une fois ingérées sont enregistrées sur MinIO au format Delta.
Delta Un format de fichier développé par Databricks. Delta est une version
évoluée du parquet qui supporte les opérations ACID..

L’ARCHITECTURE
– Les nodes sonts des machines virtuelles sur VMWare ESXi7
– Un master Spark et 8 workers [CPU 16, RAM 64GB, DISK 256GB (NVME)]
– Un node MinIO [CPU 16, RAM 64GB, DISK 256GB (NVME)]
– 6 nodes Elastic Search [CPU 16, RAM 64GB, DISK 256GB (NVME)]
LES DONNEES
Brutes: Différents gros fichiers (1G+) au format TXT, CSV
Traitées: Fichiers parquets au format delta (delta.io de Databricks)
Les données traitées contiennent un milliard d’enregistrements et une vingtaine de
colonnes d’une taille compressée (snappy) de 34-35 G0 enregistrées sur MinIO.
Le workflow Simplifié
1. Spark lit les données (fichiers delta) sur MinIO. D’une part des données
existantes, d’autre part des données avec lesquelles enrichir les données.
2. Les données existantes sont fusionnées aux données du fichier
Les données sont filtrées sur la colonne d’agrégation pour enlever les valeurs
nulles et vides. Résultant en deux jeux de données: valeurs_nulles et
valeurs_non_nulles.
3. Le jeu de données valeurs_non_nulles subit deux agrégations
4. Le jeu de données résultant de ces aggrégations valeurs_non_nulles_agg est
fusionné (unionByName) aux valeurs_nulles donnant ainsi un jeu de données
valeurs_finales
5. Le jeu de données valeurs_finales est écrit sur MinIO en écrasant les anciennes
données

LES PROBLÈMES

PROBLÈME 1
• Avec quatres workers, le temps de traitement est d’environ 30 minutes.
• Avec 8 workers, il est de 20 min environ.
• Si on se tient à ce calcul, avec 16 workers, on devrait etre à 10 – 15 minutes.
Mais on arrive comme à saturation après 10 nodes et on n’a plus de gain
significatif quand on ajoute plus de workers.

PROBLÈME 2
Les disques sont des disques NVME qui peuvent therioquement supporter une vitesse
d’écriture de l’ordre de 3,5Go/s. Le réseau est un réseau 40G, mais l’enregistrement
de 250GB de données compressées en 35 GB (snappy) sur MinIO après le traitement
dure 30 minavec 4 nodes. Ce qui semble très long vu la vitesse que peut supporter le
disque et la vitesse du réseau.