Introduction
Un script peut s’exécuter avec succès tout en transmettant de mauvais enregistrements à un rapport. La livraison de vente d’aujourd’hui contient une ligne de commande répétée, un client manquant, du texte numérique invalide, une quantité non positive, un prix négatif et une date mal formée. Remplacer automatiquement ces problèmes par zéro inventerait des informations métier.
Vous séparerez trois décisions : lire le texte sans perdre les preuves originales, convertir sans risque les champs numériques et de date prévus, puis conserver uniquement les lignes respectant une règle énoncée. Les lignes valides identiques seront ensuite dédupliquées. Enfin, vous enregistrerez et relirez un jeu de données Parquet propre. Parquet est un format de fichier binaire typé en colonnes ; ce lab enseigne uniquement le flux minimal d’écriture et de lecture, le stockage partitionné étant réservé à plus tard.
L’environnement préparé fournit deux livraisons indépendantes. Vous devez connaître la lecture CSV, les expressions de colonnes Spark, le filtrage et les prix décimaux. L’expression de conversion sûre utilisée ici est prise en charge par la version Spark préparée ; vous n’avez pas besoin de changer sa configuration de gestion des erreurs. Utilisez le Terminal. Aucun fichier de sortie précédent, Hadoop, Hive ou installation de service n’est nécessaire.
Préserver la livraison dans une zone de préparation en chaînes
Dans cette étape, vous observerez les défauts avant de convertir les valeurs. Examinez la source locale :
cd /home/labex/project
cat /home/labex/project/data/dirty-orders.csv
Il y a quatorze lignes métier. Un DataFrame de préparation (staging) conserve ces champs d’entrée sous forme de chaînes, ce qui permet de voir des valeurs comme two, unknown et not-a-date sans prétendre qu’elles sont des nombres ou des dates valides. Le comportement par défaut du lecteur CSV transforme un champ client vide en NULL ; NULL représente une valeur manquante, et non le texte « NULL ».
Enregistrez et exécutez le programme complet de préparation. Comme précédemment, le here-document PYTHON entre apostrophes écrit littéralement la source. show(30, truncate=False) autorise l’affichage de trente enregistrements au maximum et conserve le texte des champs lisible :
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
staging.printSchema()
staging.orderBy("order_id").show(30, truncate=False)
print("Staging rows:", staging.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/staging-orders.txt
cat /home/labex/project/output/staging-orders.txt
Vous devez obtenir quatorze enregistrements affichés et Staging rows: 14. Toutes les colonnes doivent être des chaînes. Repérez la quantité two de b2 et le prix unknown de b4 ; ce sont des preuves utiles, plutôt que des erreurs cachées par une valeur par défaut devinée. o2 apparaît deux fois, et le statut de o1 est entouré d’espaces.
Le fichier original est en lecture seule dans cet exercice. Vous créerez des DataFrames dérivés et des répertoires de sortie tout en laissant la livraison brute disponible pour révision.
Convertir sans risque sans inventer de valeurs par défaut
Dans cette étape, vous normaliserez les deux champs textuels dont les espaces aux extrémités doivent être retirés et tenterez les conversions numériques et de date. trim supprime les espaces au début et à la fin. Il ne devine pas un client manquant et ne change pas le statut de paiement d’une commande.
expr permet de fournir une expression intégrée de Spark sous forme de texte. try_cast(quantity AS INT) tente une conversion en entier et produit NULL si cette valeur ne peut pas être convertie. Les expressions équivalentes demandent DECIMAL(10,2) pour les prix et DATE pour les dates. Des conversions ordinaires qui échouent peuvent lever des erreurs avec le comportement ANSI de Spark ; cette conversion sûre explicite évite de dépendre d’un paramètre global.
Le changement de schéma conserve les lignes pour que les problèmes de conversion restent observables. Exécutez cette version :
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
parsed.printSchema()
parsed.orderBy("order_id").show(30, truncate=False)
print("Parsed rows:", parsed.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/parsed-orders.txt
cat /home/labex/project/output/parsed-orders.txt
Vous devez obtenir Parsed rows: 14. La quantité de b2, le prix de b4 et la date de b6 doivent maintenant être NULL. La quantité zéro de b3 et le prix négatif de b5 sont des valeurs numériques syntaxiquement valides, mais ils ne respectent toujours pas les règles métier que vous appliquerez ensuite. Le statut de o1 doit maintenant être paid sans espaces autour.
Une conversion réussie ne signifie pas qu’une commande est valide. Garder ces étapes distinctes permet d’expliquer si une ligne a échoué à la conversion ou à une contrainte métier. Aucun prix manquant n’a été remplacé par zéro ; la véritable commande gratuite o7 reste distincte d’un prix inconnu.
Appliquer des règles explicites de validité et de doublons
Dans cette étape, vous conserverez les enregistrements métier valides et supprimerez uniquement les doublons identiques. Utilisez ces règles :
- customer_id est présent et non vide après retrait des espaces ;
- quantity a été convertie avec succès et est supérieure à zéro ;
- unit_price a été converti avec succès et est supérieur ou égal à zéro ;
- order_date a été convertie avec succès ;
- les lignes identiques sur les sept champs normalisés sont conservées une seule fois.
Les autres champs d’identifiant de l’exercice sont déjà bien formés. Il ne contient pas de versions contradictoires du même identifiant de commande ; cet exercice ne choisit donc pas de version gagnante entre des statuts changeants. Les enregistrements payés, en attente et remboursés restent tous s’ils sont valides par ailleurs ; nettoyer une livraison diffère de sélectionner les revenus payés d’un rapport.
isNotNull vérifie explicitement la présence. Une comparaison impliquant NULL ne devient pas une correspondance vraie, et filter ne conserve que les conditions vraies. Combiner les conditions avec & rend la politique visible. dropDuplicates() sans sous-ensemble supprime les lignes complètes identiques ; utiliser uniquement order_id serait une autre politique, susceptible d’écarter des informations contradictoires.
Exécutez cette version :
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
col("customer_id").isNotNull()
& (col("customer_id") != "")
& col("quantity").isNotNull() & (col("quantity") > 0)
& col("unit_price").isNotNull() & (col("unit_price") >= 0)
& col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
print("Clean rows:")
clean.orderBy("order_id").show(30, truncate=False)
print("Staging count:", staging.count())
print("Valid count before deduplication:", valid.count())
print("Clean count:", clean.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/cleaning-result.txt
cat /home/labex/project/output/cleaning-result.txt
Vous devez obtenir Staging count: 14, Valid count before deduplication: 8 et Clean count: 7. b1–b6 sont exclues par les règles énoncées, tandis qu’une répétition de o2 est supprimée. Les sept identifiants conservés vont de o1 à o7.
Le véritable prix zéro de o7 est valide selon la règle de prix non négatif. o3 est toujours en attente et o5 remboursée : aucun de ces statuts ne rend l’enregistrement source mal formé. Le jeu de données obtenu est propre pour cette politique énoncée ; cela ne prétend pas résoudre toutes les anomalies possibles du monde réel.
Enregistrer, relire et tester un défaut supplémentaire
Dans cette étape, vous rendrez le jeu de données propre réutilisable et testerez la même politique sur une autre livraison. write.mode("overwrite").parquet(path) écrit un jeu de données Parquet dans un répertoire en y remplaçant le contenu dérivé précédent. Gardez ce répertoire séparé des entrées brutes. Spark peut créer plusieurs fichiers part et un marqueur de fin ; ne dépendez pas d’un nom ou nombre de fichiers précis.
spark.read.parquet(path) lit le jeu de données enregistré et ses types stockés. Interroger ce DataFrame relu vérifie le véritable résultat, plutôt que d’afficher simplement le plan avant écriture. Le programme complet ci-dessous place la logique déjà apprise dans une fonction Python pour que les deux livraisons utilisent la même règle de nettoyage.
def clean_orders(input_path): définit la fonction et nomme son paramètre d’entrée. Les lignes indentées sont son corps ; elles s’exécutent lorsque la fonction est appelée. Le premier appel fournit le nom du fichier original, et le second celui de la livraison supplémentaire. Dans chaque appel, input_path désigne la valeur fournie. return clean renvoie le DataFrame obtenu à l’appelant ; il n’enregistre pas de fichiers. Les lignes d’écriture non indentées réalisent cette action séparée. Préservez l’indentation en copiant le programme complet.
La livraison supplémentaire contient les quatorze lignes originales, plus une ligne de quantité négative et une autre copie exacte de la commande valide à prix zéro. Prévoyez le résultat propre : ces ajouts ne doivent pas changer les sept enregistrements conservés. Examinez les ajouts :
tail -n 2 /home/labex/project/data/dirty-orders-extra.csv
Exécutez le programme complet d’enregistrement et de relecture :
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
def clean_orders(input_path):
staging = spark.read.option("header", True).csv(input_path)
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
col("customer_id").isNotNull()
& (col("customer_id") != "")
& col("quantity").isNotNull() & (col("quantity") > 0)
& col("unit_price").isNotNull() & (col("unit_price") >= 0)
& col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
return clean
clean = clean_orders("/home/labex/project/data/dirty-orders.csv")
clean.write.mode("overwrite").parquet("/home/labex/project/output/clean-orders")
reloaded = spark.read.parquet("/home/labex/project/output/clean-orders")
print("Read-back clean rows:")
reloaded.orderBy("order_id").show(30, truncate=False)
reloaded.printSchema()
print("Read-back count:", reloaded.count())
practice = clean_orders("/home/labex/project/data/dirty-orders-extra.csv")
practice.write.mode("overwrite").parquet("/home/labex/project/output/practice-clean-orders")
practice_reloaded = spark.read.parquet("/home/labex/project/output/practice-clean-orders")
print("Extra-delivery clean rows:")
practice_reloaded.orderBy("order_id").show(30, truncate=False)
print("Extra-delivery count:", practice_reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/clean-parquet-check.txt
cat /home/labex/project/output/clean-parquet-check.txt
Les deux tables relues doivent contenir o1 à o7 exactement une fois, et les deux nombres de lignes doivent être 7. Le répertoire de sortie principal est output/clean-orders ; practice-clean-orders est une expérience dérivée séparée. Le schéma relu doit conserver la quantité integer, le prix decimal et order_date de type date.
C’est un aller-retour Parquet minimal, pas une comparaison de vitesse ni un exercice d’optimisation du stockage. Vous disposez maintenant d’un petit jeu de données propre qu’un autre processus Spark peut lire, avec la preuve qu’une nouvelle ligne invalide et un doublon ne changent pas silencieusement le résultat.
Résumé
Vous avez préservé le texte brut, utilisé une conversion sûre pour exposer les valeurs invalides, appliqué une politique explicite de présence et de plage, puis supprimé les lignes dupliquées identiques. Vous avez conservé les prix zéro valides et les enregistrements non payés, puis vérifié la véritable sortie Parquet typée et une seconde livraison. Le prochain défi vous permettra d’appliquer de façon autonome ces compétences de nettoyage et de calcul de montants.
