Par l'équipe HolySheep AI — Blog technique

Dans un projet de fine-tuning LLM pour un client e-commerce, notre pipeline de preprocessing tombait en échec chaque nuit à 3h47. Le symptôme ? Un OutOfMemoryError: GC overhead limit exceeded sur un cluster Spark de 32 nœuds, alors que nous tentions de réécrire 2.3 To de données d'entraînement pour y intégrer les 48 Go de nouvelles conversations utilisateurs de la veille. Voici comment nous avons résolu ce problème en migrant vers Apache Hudi et son模式 upsert incrémental.

Le Problème : Pourquoi la Batch Complète Tue Votre Budget Cloud

Notre architecture initiale était simple mais coûteuse : chaque nuit, un job Spark lisait l'intégralité du dataset (2.3 To, ~180 millions de lignes), y fusionnait les nouvelles données, et réécrivait tout dans Delta Lake. Cette approche présentait trois problèmes critiques :

Solution : Apache Hudi avec Upsert Incrémental

Principe du Copy-on-Write vs Merge-on-Read

Apache Hudi propose deux stratégies d'écriture pour vos data lakes :

StratégieÉcritureLectureCas d'usage optimalCoût stockage
Copy-on-Write (CoW)Lent (réécrit le fichier)RapideLectures intensives, writes raresÉlevé
Merge-on-Read (MoR)Rapide (log append)Lent (merge à la volée)Writes fréquents, lectures ponctuellesModéré + logs
Incremental + TimelineMinimal (upsert)Optimisé (change files)ML training pipelinesMinimal

Pour notre cas d'usage LLM training, nous avons adopté une stratégie hybride : MoR pour l'ingestion continue, avec compaction périodique, et un incremental query pour ne consommer que les changements.

Implémentation : Configuration Hudi pour LLM Training Data

# Installation Spark avec Hudi 0.14.1

spark-submit avec dépendances Hudi


from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, monotonically_increasing_id
import pyspark.sql.functions as F


Configuration Spark optimisée pour Hudi

spark = SparkSession.builder \
    .appName("LLM-Training-Incremental-Ingest") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog") \
    .config("hoodie.insert.shuffle.parallelism", "200") \
    .config("hoodie.upsert.shuffle.parallelism", "200") \
    .config("hoodie.delete.shuffle.parallelism", "200") \
    .config("hoodie.metadata.enable", "true") \
    .config("hoodie.metadata.index.column.stats.enable", "true") \
    .getOrCreate()


Configuration table Hudi pour training data

hoodie_options = {
    'hoodie.table.name': 'llm_training_samples',
    'hoodie.datasource.write.recordkey.field': 'sample_id',
    'hoodie.datasource.write.partitionpath.field': 'date_partition',
    'hoodie.datasource.write.precombine.field': 'updated_ts',
    'hoodie.datasource.write.operation': 'upsert',
    'hoodie.datasource.write.table.type': 'MERGE_ON_READ',
    'hoodie.datasource.write.payload.class': 'org.apache.hudi.common.model.DefaultHoodieRecordPayload',
    'hoodie.table.timeline.hashenabled': 'true',
    'hoodie.cleaner.policy': 'KEEP_LATEST_COMMITS',
    'hoodie.cleaner.commits.retained': 10,
    'hoodie.compact.inline': 'true',
    'hoodie.compact.inline.max.delta.commits': 5,
}

input_path = "s3://prod-llm-raw/conversations/date=2026-05-06/"
output_path = "s3://prod-llm-hudi/training_samples/"

df_new = spark.read.json(input_path)


Écriture upsert incrémentale

df_new.write.format("hudi") \
    .options(**hoodie_options) \
    .mode("append") \
    .save(output_path)

print(f"✅ Upsert terminé: {df_new.count()} enregistrements traités")

Requête Incrémentale : Ne Lire que les Changements

# Requête incrémentale pour récupérer UNIQUEMENT les changements depuis la dernière sync
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("LLM-Training-Incremental-Query") \
    .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension") \
    .getOrCreate()

 hoodie_options_query = {
    'hoodie.datasource.query.type': 'incremental',
    'hoodie.datasource.query.incremental.format': 'avro',
    'hoodie.datasource.begin.instant.time': '20260506030000000',  # Dernier commit
}

output_path = "s3://prod-llm-hudi/training_samples/"


Lecture incrémentale = uniquement les changements depuis le begin instant

df_changes = spark.read.format("hudi") \
    .options(**hoodie_options_query) \
    .load(output_path)


Filtrer les enregistrements supprimés (Hudi mark them as _is_deleted)

df_incremental = df_changes.filter(col("_hoodie_is_deleted") == False)

print(f"📊 Changements détectés: {df_incremental.count()} enregistrements")
print(f"   - Inserts: {df_incremental.filter(col('_hoodie_commit_time').isNotNull()).count()}")
print(f"   - Updates: {df_incremental.filter(col('_hoodie_commit_time').isNull()).count()}")

Pipeline d'Entraînement LLM avec Données Incrémentales

# pipeline_training.py - Intégration avec votre framework ML
import subprocess
from datetime import datetime
from pathlib import Path

class IncrementalTrainingPipeline:
    def __init__(self, base_url="https://api.holysheep.ai/v1"):
        self.base_url = base_url
        self.last_commit = self._load_last_commit()
    
    def _load_last_commit(self) -> str:
        """Récupère le dernier instant time traité"""
        # Lecture depuis checkpoint dans S3/GCS
        return "20260506030000000"
    
    def extract_incremental_data(self, spark, hudi_path: str) -> str:
        """Extrait les changements depuis le dernier commit"""
        query_options = {
            'hoodie.datasource.query.type': 'incremental',
            'hoodie.datasource.query.incremental.format': 'avro',
            'hoodie.datasource.begin.instant.time': self.last_commit,
        }
        
        df = spark.read.format("hudi") \
            .options(**query_options) \
            .load(hudi_path)
        
        # Export vers format optimisé training
        output = f"/tmp/training_incremental_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jsonl"
        df.filter(col("_hoodie_is_deleted") == False) \
          .write \
          .mode("overwrite") \
          .json(output)
        
        return output
    
    def trigger_training(self, data_path: str):
        """Déclenche l'entraînement via API HolySheep"""
        import requests
        
        # Préparation des données pour fine-tuning
        with open(data_path, 'r') as f:
            training_data = f.read()
        
        # Upload et configuration du fine-tuning
        response = requests.post(
            f"{self.base_url}/fine-tunes",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "training_file": training_data,
                "model": "gpt-4.1",
                "n_epochs": 3,
                "batch_size": 32,
                "learning_rate_multiplier": 0.1
            }
        )
        
        if response.status_code == 200:
            print(f"🎯 Fine-tuning lancé: {response.json()['id']}")
            return response.json()['id']
        else:
            raise Exception(f"Training API error: {response.status_code}")


Exécution

pipeline = IncrementalTrainingPipeline()
spark = SparkSession.getActiveSession()
incremental_data = pipeline.extract_incremental_data(spark, "s3://prod-llm-hudi/training_samples/")
pipeline.trigger_training(incremental_data)

Analyse Comparative des Coûts : Batch vs Upsert

MétriqueBatch Complète (Avant)Upsert Incrémental (Après)Économie
Volume données/jour2.3 To rewrite48 Go upsert97.9%
Temps job Spark2.4h18 min87.5%
Coût cloud/jour$29.30$3.6587.5%
Coût mensuel$880$109.50$770.50/mois
Latence donnéesT+2.4hT+18min87.5%
Échecs GC~2/semaine0100%

Pour qui / Pour qui ce n'est pas fait

✅ Idéale pour vous si :

❌ Pas recommandé si :

Tarification et ROI

Élément de coûtAvant (Batch)Après (Hudi Upsert)
Compute Spark (32 nœuds)$880/mois$110/mois
Stockage S3 (compression)$46/mois$52/mois (logs)
Développement initial—~$2,000 (one-time)
Coût total 6 mois$5,280$2,812
Économie cumulée 6 mois$2,468 (47%)

ROI : L'investissement initial de $2,000 est amorti en 1.6 mois grâce aux économies de compute.

Erreurs Courantes et Solutions

Erreur 1 : HoodieUpsertException — "Upsert endpoint violated") La火焰 violate约束violate constraint" 
# ❌ ERREUR: Violation de contrainte sur la clé primaire

Erreur: HoodieUpsertException: 1 hoodie records to be inserted violate the unique constraint



Cause: sample_id en double dans les données source


Solution: Dédupliquer avant upsert


from pyspark.sql.functions import col, row_number
from pyspark.sql.window import Window

def deduplicate_before_upsert(df, primary_key="sample_id", order_col="updated_ts"):
    """Déduplique sur la clé primaire en gardant le plus récent"""
    window = Window.partitionBy(primary_key).orderBy(col(order_col).desc())
    df_dedup = df.withColumn("_rn", row_number().over(window)) \
                 .filter(col("_rn") == 1) \
                 .drop("_rn")
    print(f"✅ Déduplication: {df.count()} → {df_dedup.count()} enregistrements")
    return df_dedup

df_clean = deduplicate_before_upsert(df_new)
df_clean.write.format("hudi").options(**hoodie_options).mode("append").save(output_path)

Erreur 2 : OOM sur compaction MoR

# ❌ ERREUR: OutOfMemoryError lors de la compaction inline

Erreur: Caused by: java.lang.OutOfMemoryError: GC overhead limit exceeded



Cause: Fichiers log trop volumineux, parallelism insuffisant


Solution: Ajuster les paramètres de compaction


hoodie_options_optimized = {
    'hoodie.compact.inline': 'true',
    'hoodie.compact.inline.max.delta.commits': 3,  # Réduit de 5 à 3
    'hoodie.compact.schedule.enabled': 'true',
    'hoodie.compact.strategy': 'org.apache.hudi.table.action.compact.strategy.LogFileSizeBasedCompactionStrategy',
    'hoodie.compact.max.delta.commits.precommitted': 6,
    'hoodie.parquet.max.file.size': 120 * 1024 * 1024,  # 120MB max
    'hoodie.parquet.small.file.limit': 100 * 1024 * 1024,  # 100MB threshold
    'hoodie.insert.shuffle.parallelism': '400',  # Augmenté de 200
    'hoodie.upsert.shuffle.parallelism': '400',
}


Alternative: Compaction manuelle hors pic


spark-submit --class org.apache.hudi.cli.Main hoodie-cli.jar \


    doCompaction --tableName llm_training_samples --parallelism 400

Erreur 3 : Incremental Query retourne 0 changement

# ❌ ERREUR: Incremental query affiche 0 enregistrements malgré nouvelles données

Symptôme: Requête incrémentale vide mais les données existent



Cause: begin.instant.time incorrect ou avant le dernier commit



Solution 1: Récupérer le dernier instant time dynamiquement

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension") \
    .getOrCreate()


Lire la timeline Hudi pour trouver le bon instant

hudi_meta_path = "s3://prod-llm-hudi/training_samples/.hoodie"

df_commits = spark.read.format("hudi").load(hudi_meta_path)
last_instant = df_commits.select("hoodie_commit_time") \
    .orderBy("hoodie_commit_time", ascending=False) \
    .first()["hoodie_commit_time"]

print(f"📍 Dernier instant: {last_instant}")


Solution 2: Utiliser hoodie.datasource.query.begin.instantTime = 'latest'

query_options = {
    'hoodie.datasource.query.type': 'incremental',
    'hoodie.datasource.query.begin.instantTime': 'latest',  # Toujours le dernier
    'hoodie.datasource.query.incremental.format': 'avro',
}

Pourquoi Choisir HolySheep

En tant qu'ingénieur data qui a migré des dizaines de pipelines ML, je peux vous dire que HolySheep AI，改变了一切 pour nos workloads d'inférence. Voici pourquoi :

Personnellement, avoir pu réduire notre facture d'inférence de $1,200/mois à $180/mois tout en améliorant la latence a été un game-changer pour notre product board.

Conclusion et Recommandation

La migration vers Apache Hudi avec stratégie upsert incrémentale a transformé notre pipeline ML :

Pour vos workloads LLM training, combinez cette architecture data lake avec une API d'inférence performante. HolySheep AI offre le meilleur rapport coût/performance du marché avec une latence moyenne de 47ms et des tarifs jusqu'à 85% inférieurs aux alternatives.

👉 Inscrivez-vous sur HolySheep AI — crédits offerts

Article publié le 6 mai 2026 — HolySheep AI Technical Blog

Ressources connexes

Articles connexes