src.pipeline.aggregator

aggregator.py — Module d'agrégation des sinistres par client

 1"""
 2aggregator.py — Module d'agrégation des sinistres par client
 3"""
 4
 5from pyspark.sql import DataFrame
 6from pyspark.sql.functions import avg, count
 7from pyspark.sql.functions import sum as _sum
 8
 9
10def aggregate_data(df: DataFrame) -> DataFrame:
11    """
12    Agrège les données par client :
13    - total des sinistres
14    - moyenne des montants
15    - nombre de sinistres
16
17    Args:
18        df (DataFrame): DataFrame Spark nettoyé contenant les colonnes 'client_id' et 'montant_clean'
19
20    Returns:
21        DataFrame: DataFrame agrégé par client
22    """
23    return df.groupBy("client_id").agg(
24        _sum("montant_clean").alias("total"),
25        avg("montant_clean").alias("moyenne"),
26        count("*").alias("nb_sinistres"),
27    )
def aggregate_data(df: pyspark.sql.dataframe.DataFrame) -> pyspark.sql.dataframe.DataFrame: 
11def aggregate_data(df: DataFrame) -> DataFrame:
12    """
13    Agrège les données par client :
14    - total des sinistres
15    - moyenne des montants
16    - nombre de sinistres
17
18    Args:
19        df (DataFrame): DataFrame Spark nettoyé contenant les colonnes 'client_id' et 'montant_clean'
20
21    Returns:
22        DataFrame: DataFrame agrégé par client
23    """
24    return df.groupBy("client_id").agg(
25        _sum("montant_clean").alias("total"),
26        avg("montant_clean").alias("moyenne"),
27        count("*").alias("nb_sinistres"),
28    )

Agrège les données par client :

  • total des sinistres
  • moyenne des montants
  • nombre de sinistres

Args: df (DataFrame): DataFrame Spark nettoyé contenant les colonnes 'client_id' et 'montant_clean'

Returns: DataFrame: DataFrame agrégé par client