src.pipeline.cleaner

 1"""
 2cleaner.py — Nettoyage des données sinistres pour le pipeline PySpark
 3
 4Contient des fonctions de nettoyage métier :
 5- montant négatif ou null → remplacé par la médiane
 6- age hors bornes → remplacé par défaut
 7- date future ou null → remplacée par référence
 8- type de sinistre invalide → remplacé par 'Autre'
 9"""
10
11from pyspark.sql import DataFrame
12from pyspark.sql.functions import col, lit, when
13
14
15def nettoyer_montant(df: DataFrame, mediane: int) -> DataFrame:
16    return df.withColumn(
17        "montant_clean", when((col("montant") < 0) | col("montant").isNull(), lit(mediane)).otherwise(col("montant"))
18    )
19
20
21def nettoyer_age(df: DataFrame, age_defaut=18, age_min=18, age_max=80) -> DataFrame:
22    return df.withColumn(
23        "age_clean",
24        when((col("age") < age_min) | (col("age") > age_max) | col("age").isNull(), lit(age_defaut)).otherwise(
25            col("age")
26        ),
27    )
28
29
30def nettoyer_date(df: DataFrame, date_reference: str) -> DataFrame:
31    return df.withColumn(
32        "date_sinistre_clean",
33        when(col("date_sinistre") > lit(date_reference) | col("date_sinistre").isNull(), lit(date_reference)).otherwise(
34            col("date_sinistre")
35        ),
36    )
37
38
39def nettoyer_type_sinistre(df: DataFrame, types_valides, type_defaut="Autre") -> DataFrame:
40    return df.withColumn(
41        "type_sinistre_clean",
42        when(col("type_sinistre").isin(types_valides), col("type_sinistre")).otherwise(lit(type_defaut)),
43    )
44
45
46def clean_dataframe(df: DataFrame) -> DataFrame:
47    """
48    Nettoie le DataFrame brut en appliquant les règles métier :
49    - montant → montant_clean
50    - age → age_clean
51    - date_sinistre → date_sinistre_clean
52    - type_sinistre → type_sinistre_clean
53
54    Args:
55        df (DataFrame): DataFrame brut
56
57    Returns:
58        DataFrame: DataFrame nettoyé
59    """
60    mediane = 8000
61    types_valides = ["vol", "accident", "bris de glace", "incendie", "catastrophe naturelle"]
62
63    df = nettoyer_montant(df, mediane)
64    df = nettoyer_type_sinistre(df, types_valides)
65
66    return df