Nettoyage des données avec Apache Spark en Python
À l'heure actuelle, où le volume et la vitesse des flux de données sont élevés, Apache Spark, un framework de traitement de Big Data open source, est un choix courant car il permet un traitement parallèle et distribué des données. Le nettoyage de ces données est une étape importante et Apache Spark nous fournit une variété d'outils et de méthodes pour le nettoyage des données. Dans cette méthode, nous allons voir comment nettoyer les données avec Apache Spark en Python et les étapes pour ce faire sont les suivantes :
Chargement des données dans un Spark DataFrame - La méthode SparkSession.read nous permet de lire des données provenant de diverses sources comme CSV, JSON, Parquet, etc.
Gestion des valeurs manquantes - Les méthodes DataFrame.dropna ou DataFrame.fillna nous permettent de supprimer les données avec toutes les valeurs manquantes ou de remplir les valeurs manquantes avec une valeur particulière respectivement.
Gérer les doublons – Les données contiennent souvent des entrées en double. Pour gérer cela, la méthode DataFrame.dropDuplicates nous permet de supprimer les doublons du DataFrame.
Gestion des valeurs aberrantes - La méthode DataFrame.filter nous permet de supprimer toutes les lignes qui contiennent des valeurs aberrantes.
-
Gestion des valeurs aberrantes - Pour convertir les types de données des colonnes, nous avons une méthode appelée DataFrame.cast
Avant d'aller de l'avant et de voir comment nettoyer les données avec pyspark, nous devrons obligatoirement installer la bibliothèque pyspark. Pour ce faire, nous devons exécuter la commande suivante dans le terminal :
pip install pysparkGestion des valeurs manquantes
La gestion des valeurs manquantes dans Apache Spark implique d'identifier et de traiter les données manquantes ou incomplètes dans un ensemble de données stocké dans un Spark DataFrame. Il existe plusieurs méthodes pour gérer les valeurs manquantes dans Spark, notamment :
Suppression de lignes - Cela implique la suppression des enregistrements contenant des valeurs manquantes du DataFrame.
Imputer les valeurs manquantes - Cela implique de remplacer les valeurs manquantes par une valeur calculée telle que la moyenne, la médiane ou le mode des données dans la colonne.
Remplir les valeurs manquantes - Cela implique de remplacer les valeurs manquantes par une valeur spécifique, telle que zéro, ou une valeur par défaut.
Interpoler les valeurs manquantes - Cela implique l'utilisation de méthodes mathématiques telles que l'interpolation linéaire ou l'interpolation spline pour estimer les valeurs manquantes.
La méthode utilisée pour gérer les valeurs manquantes dépend des exigences spécifiques et des objectifs de l'analyse des données. Il est important de traiter les valeurs manquantes de manière cohérente et reproductible pour garantir l'intégrité des données et l'exactitude des résultats.
Dans Apache Spark, les fonctions fournies par les modules pyspark.sql.DataFrame et pyspark.sql.DataFrameNaFunctions peuvent être utilisées pour gérer les valeurs manquantes. Ces fonctions incluent dropna, fillna et interpolate.
Exemple
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("MissingData").getOrCreate()
# Create a sample data frame
data = [("John", 25, None), ("Jane", 30, 35.5), ("Jim", None, 40.0), ("Joan", 32, None)]
columns = ["Name", "Age", "Salary"]
df = spark.createDataFrame(data, columns)
# Display the original data frame
print("Original Data Frame:")
df.show()
# Replacing Missing Values with Mean
from pyspark.sql.functions import mean
mean_age = df.agg(mean(df["Age"])).first()[0]
mean_salary = df.agg(mean(df["Salary"])).first()[0]
df = df.fillna({"Age": mean_age, "Salary": mean_salary})
# Display the cleaned data frame
print("Cleaned Data Frame:")
df.show()
spark.stop()Sortir
Original Data Frame:
+----+----+------+
|Name| Age|Salary|
+----+----+------+
|John| 25| null|
|Jane| 30| 35.5|
| Jim|null| 40.0|
|Joan| 32| null|
+----+----+------+
Cleaned Data Frame:
+----+---+------+
|Name|Age|Salary|
+----+---+------+
|John| 25| 37.75|
|Jane| 30| 35.5|
| Jim| 29| 40.0|
|Joan| 32| 37.75|
+----+---+------+Gestion des doublons
La gestion des doublons dans Apache Spark implique l'identification et le traitement des enregistrements en double dans un ensemble de données stocké dans un Spark DataFrame. Il existe plusieurs méthodes pour gérer les doublons dans Spark, notamment :
-
Suppression des doublons - Cela implique d'identifier et de supprimer les enregistrements en double du DataFrame. La fonction dropDuplicates peut être utilisée pour supprimer les enregistrements en double dans Spark.
Conserver les doublons - Cela implique de conserver toutes les instances d'enregistrements en double dans le DataFrame, généralement en ajoutant un identifiant ou un index unique à chaque enregistrement.
Marquage des doublons - Cela implique de marquer les enregistrements en double dans le DataFrame, mais pas de les supprimer, afin qu'ils puissent être analysés ou traités davantage.
La méthode utilisée pour gérer les doublons dépend des exigences spécifiques et des objectifs de l'analyse des données. Il est important de traiter les doublons de manière cohérente et reproductible pour garantir l’intégrité des données et l’exactitude des résultats.
Dans Apache Spark, la fonction dropDuplicates peut être utilisée pour supprimer les enregistrements en double dans un DataFrame. La fonction prend une ou plusieurs colonnes en entrée et supprime tous les enregistrements dont les valeurs dans les colonnes spécifiées sont identiques. La fonction dropDuplicates renvoie un nouveau DataFrame avec les enregistrements en double supprimés.
Exemple
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("DuplicateData").getOrCreate()
# Create a sample data frame
data = [("John", 25, 90.0), ("Jane", 30, 35.5), ("Jim", 20, 200.0), ("Joan", 32, 50.0),
("John", 25, 90.0), ("Jim", 20, 200.0)]
columns = ["Name", "Age", "Salary"]
df = spark.createDataFrame(data, columns)
# Display the original data frame
print("Original Data Frame:")
df.show()
# Remove duplicates
df = df.dropDuplicates()
# Display the cleaned data frame
print("Cleaned Data Frame:")
df.show()
spark.stop()Sortir
Original Data Frame:
+----+---+------+
|Name|Age|Salary|
+----+---+------+
|John| 25| 90.0|
|Jane| 30| 35.5|
| Jim| 20| 200.0|
|Joan| 32| 50.0|
|John| 25| 90.0|
| Jim| 20| 200.0|
+----+---+------+
Cleaned Data Frame:
+----+---+------+
|Name|Age|Salary|
+----+---+------+
|Jane| 30| 35.5|
|John| 25| 90.0|
| Jim| 20| 200.0|
|Joan| 32| 50.0|
+----+---+------+Gestion des valeurs aberrantes
La gestion des valeurs aberrantes dans Apache Spark fait référence au processus d'identification et de suppression ou de transformation des valeurs dans un ensemble de données qui sont considérées comme extrêmes ou en dehors de la plage normale de valeurs. Les valeurs aberrantes peuvent avoir un impact significatif sur les résultats des analyses statistiques, il est donc souvent nécessaire de les traiter d'une manière ou d'une autre.
Il existe plusieurs approches courantes pour gérer les valeurs aberrantes dans Apache Spark, notamment :
Suppression des enregistrements contenant des valeurs aberrantes : cela implique de filtrer les enregistrements dans lesquels la valeur d'une colonne particulière se situe en dehors d'une plage spécifiée ou se situe en dehors d'un certain nombre d'écarts types par rapport à la moyenne.
Remplacer les valeurs aberrantes par la moyenne ou la médiane - Cela implique de remplacer les valeurs considérées comme des valeurs aberrantes par la moyenne ou la médiane des valeurs restantes dans la colonne.
Winsoriser les valeurs aberrantes - Cela implique de remplacer les valeurs qui sont des valeurs aberrantes par une valeur de centile spécifiée, telle que le 5e ou le 95e centile.
-
Clipping outliers - Cela implique de remplacer les valeurs qui sont des valeurs aberrantes par une valeur maximale ou minimale spécifiée.
Pour gérer les valeurs aberrantes dans Apache Spark, vous pouvez utiliser les fonctions intégrées disponibles dans le module pyspark.sql.functions pour calculer des statistiques telles que la moyenne et l'écart type, puis utiliser les méthodes filter ou withColumn pour supprimer ou remplacer les valeurs aberrantes comme vous le souhaitez. .
Exemple
from pyspark.sql import SparkSession
from pyspark.sql.functions import mean, stddev, abs
spark = SparkSession.builder.appName("OutlierHandlingExample").getOrCreate()
# Create a sample data frame
data = [("John", 25, 90.0), ("Jane", 30, 35.5), ("Jim", 20, 200.0), ("Joan", 32, 50.0)]
columns = ["Name", "Age", "Salary"]
df = spark.createDataFrame(data, columns)
# Display the original data frame
print("Original Data Frame:")
df.show()
# Calculate mean and standard deviation
mean_salary = df.agg(mean(df["Salary"])).first()[0]
stddev_salary = df.agg(stddev(df["Salary"])).first()[0]
# Identify and filter out outliers
df = df.filter(abs(df["Salary"] - mean_salary) < stddev_salary)
# Display the cleaned data frame
print("Cleaned Data Frame:")
df.show()
spark.stop()Sortir
Original Data Frame:
+----+---+------+
|Name|Age|Salary|
+----+---+------+
|John| 25| 90.0|
|Jane| 30| 35.5|
| Jim| 20| 200.0|
|Joan| 32| 50.0|
+----+---+------+
Cleaned Data Frame:
+----+---+------+
|Name|Age|Salary|
+----+---+------+
|John| 25| 90.0|
|Jane| 30| 35.5|
|Joan| 32| 50.0|
+----+---+------+Conversion des types de données
La conversion de types de données fait référence au processus de modification de la représentation des données d'un type de données à un autre. Lors du traitement et de l’analyse des données, il est courant de rencontrer des données dans des formats différents qui ne conviennent pas à l’analyse souhaitée. Dans de tels cas, la conversion du type de données dans un format approprié est nécessaire pour effectuer correctement l'analyse.
Par exemple, dans un bloc de données, une colonne peut avoir le type de données chaîne, mais les valeurs de cette colonne sont des nombres. Dans ce cas, le type de données de la colonne doit être modifié en entier ou en nombre flottant, en fonction des exigences de l'analyse. De même, une colonne peut avoir le type de données entier, mais les valeurs de cette colonne sont des chaînes de date. Dans ce cas, le type de données de la colonne doit être modifié en type date.
La conversion des types de données est une étape importante dans le nettoyage et le prétraitement des données, car elle garantit que les données sont au format correct pour l'analyse.
Exemple
from pyspark.sql import SparkSession
from pyspark.sql.types import IntegerType, FloatType
spark = SparkSession.builder.appName("DataTypeConversion").getOrCreate()
# Create a sample data frame
data = [("John", "25", "90"), ("Jane", "30", "35"), ("Jim", "20", "200"), ("Joan", "32", "50")]
columns = ["Name", "Age", "Salary"]
df = spark.createDataFrame(data, columns)
# Display the original data frame
print("Original Data Frame:")
df.show()
# Convert the data type of the 'Age' column to integer
df = df.withColumn("Age", df["Age"].cast(IntegerType()))
# Convert the data type of the 'Salary' column to float
df = df.withColumn("Salary", df["Salary"].cast(FloatType()))
# Display the converted data frame
print("Converted Data Frame:")
df.show()
spark.stop()Sortir
Original Data Frame:
+----+---+------+
|Name|Age|Salary|
+----+---+------+
|John| 25| 90|
|Jane| 30| 35|
| Jim| 20| 200|
|Joan| 32| 50|
+----+---+------+
Converted Data Frame:
+----+---+------+
|Name|Age|Salary|
+----+---+------+
|John| 25| 90.0|
|Jane| 30| 35.0|
| Jim| 20| 200.0|
|Joan| 32| 50.0|
+----+---+------+Conclusion
Le nettoyage des données dans Apache Spark est une partie essentielle du processus de préparation des données. Apache Spark nous fournit une plate-forme puissante et efficace pour traiter de grands ensembles de données et nous aider à effectuer diverses tâches de nettoyage de données telles que la gestion des valeurs manquantes, des doublons, etc. en même temps. Le module pyspark.sql.functions nous fournit un grand nombre de fonctions qui, combinées à la capacité d'effectuer des opérations complexes dans un environnement distribué, font d'Apache Spark un choix parfait pour le nettoyage et la préparation des données. En utilisant les bons outils et techniques, nous pouvons préparer nos données pour l'analyse, le ML ou tout autre type d'application basée sur les données avec une plus grande chance d'obtenir des résultats précis.