Introduction
Le commerçant dispose déjà d’un ensemble correct et typé de résultats de commandes payées. Une autre application doit maintenant les lire plus tard, sélectionner une date et pouvoir compter sur l’absence de duplication silencieuse des commandes après un export répété. Enregistrer les données fait partie du contrat de traitement, et pas seulement de la dernière ligne d’un script.
Vous écrirez du Parquet avec un partitionnement en répertoires par order_date. La racine du jeu de données contiendra des sous-répertoires nommés par date, et Spark reconstruira la colonne de date lors de la lecture de cette racine. Ces répertoires de stockage diffèrent des partitions de calcul utilisées pour exécuter le travail ; le prochain lab examinera l’exécution séparément.
Ce lab prépare indépendamment quatre commandes payées avec des montants décimaux dans une entrée Parquet locale. Vous devez connaître les lectures DataFrame, les filtres et le flux minimal d’écrasement Parquet. Aucun catalogue Hive, HDFS, prérequis SQL, installation ou sortie précédente n’est nécessaire. Utilisez le Terminal.
Écrire un jeu de données partitionné par date et le relire
Dans cette étape, vous organiserez les résultats préparés dans des répertoires de dates. Le jeu de données d’entrée sous data/prepared-orders contient quatre commandes, avec les mêmes types de quantité, prix décimal, date et order_amount décimal utilisés précédemment. Parquet est binaire ; examinez-le avec Spark plutôt que d’attendre de cat l’affichage de texte CSV.
Commencez dans votre espace de travail :
cd /home/labex/project
partitionBy("order_date") indique à l’écriture de placer les lignes dans des répertoires nommés par date. mode("overwrite") remplace le jeu de données dérivé existant au chemin de sortie indiqué. Gardez ce chemin de sortie séparé d’input_path, qui doit rester intact.
Enregistrez et exécutez le programme complet :
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/partitioned-read-back.txt
cat /home/labex/project/output/partitioned-read-back.txt
Vous devez obtenir quatre enregistrements reconstruits : o1 et o2 le 14 octobre, o4 et o6 le 15 octobre, avec les montants 20.00, 25.00, 50.00 et 30.00. Main rows doit être 4. Vérifiez que quantity est integer, le prix et le montant decimal, et order_date date.
Une colonne de partition peut apparaître à une position différente après lecture, car Spark la reconstruit depuis les noms de répertoires plutôt que de la stocker comme champ ordinaire dans chaque fichier de partition. Les noms et les valeurs des colonnes établissent leur identité ; utilisez un select explicite si un consommateur ultérieur exige un ordre de colonnes précis. L’ordre d’affichage nécessite aussi orderBy.
Examiner les répertoires et sélectionner une date
Dans cette étape, vous relierez l’organisation des répertoires à la colonne de date reconstruite. Listez la sortie récursivement ; -R inclut les sous-répertoires de dates et leurs fichiers :
ls -R /home/labex/project/output/orders-by-date
Vous devez trouver les répertoires order_date=2026-10-14 et order_date=2026-10-15. Les noms de fichiers part, leur nombre et les métadonnées cachées peuvent varier ; ne codez pas en dur un fichier part comme s’il était le jeu de données. La racine pertinente est orders-by-date.
Lire cette racine permet à Spark de découvrir les valeurs de partition. Cette découverte de l’organisation des fichiers ne nécessite pas d’enregistrer une partition ou une table Hive. Le script indépendant suivant lit la racine, compare sa colonne DateType à une chaîne de date ISO valide et enregistre séparément les enregistrements du jour sélectionné :
cat > /home/labex/project/inspect_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("InspectDateDataset").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
root = "/home/labex/project/output/orders-by-date"
df = spark.read.parquet(root)
selected = df.filter(col("order_date") == "2026-10-14")
selected.orderBy("order_id").show(truncate=False)
selected.write.mode("overwrite").parquet("/home/labex/project/output/selected-day")
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_sales_data.py > /home/labex/project/output/selected-date.txt
cat /home/labex/project/output/selected-date.txt
Vous devez obtenir uniquement o1 et o2, avec leurs dates du 14 octobre et les montants 20.00 et 25.00. L’autre date existe toujours dans le jeu de données principal. Une lecture filtrée change les enregistrements renvoyés ; elle ne supprime pas les répertoires non sélectionnés.
Vous avez examiné l’organisation du stockage et le véritable résultat de requête. Sur ce petit échantillon, le temps écoulé ne prouve pas qu’un format ou une organisation est plus rapide. Le but est un jeu de données correct et découvrable que des lecteurs ultérieurs peuvent utiliser.
Comparer sans risque l’écrasement répété à l’ajout
Dans cette étape, vous préserverez le jeu de données principal correct et expérimenterez dans un répertoire séparé. Prévoyez les nombres de lignes jetables : écrire deux fois les quatre mêmes lignes avec écrasement laisse quatre lignes à chaque fois ; ajouter ensuite une copie complète produit huit lignes. Ces lignes supplémentaires sont des exports dupliqués, et non de nouvelles commandes.
Exécutez cette version complète. Le chemin principal utilise l’écrasement. Seul write-mode-demo reçoit l’ajout volontaire :
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
practice_path = "/home/labex/project/output/write-mode-demo"
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("First overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("Second overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("append").partitionBy("order_date").parquet(practice_path)
print("After append rows:", spark.read.parquet(practice_path).count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
Repérez Main rows: 4, First overwrite rows: 4, Second overwrite rows: 4 et After append rows: 8. Le jeu de données jetable contient maintenant volontairement deux copies de chaque commande, montrant pourquoi ajouter un instantané complet ne convient pas à l’export principal.
Réexécutez le programme enregistré pour vérifier la reproductibilité du flux complet :
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
Les mêmes nombres doivent apparaître. Chaque exécution complète reconstruit le jeu de données principal et réinitialise l’expérience jetable avant son ajout volontaire ; elle ne grandit donc pas jusqu’à douze ou seize lignes. L’ajout est une politique distincte pour des données véritablement nouvelles sans chevauchement ; l’écrasement est la politique choisie pour cet instantané dérivé complet.
Le jeu de données original prepared-orders reste disponible. Les répertoires de dates décrivent des valeurs stockées, tandis que les partitions de calcul à l’exécution de Spark décrivent la division du travail d’un traitement. Vous relierez ensuite ce second sens de partition aux plans et à l’exécution réelle.
Résumé
Vous avez écrit un jeu de données Parquet partitionné par date, reconstruit ses enregistrements typés depuis la racine et sélectionné une date sans retirer les autres. L’écrasement répété a préservé l’instantané principal, tandis qu’un ajout jetable a montré les copies en double. Les plans d’exécution et les événements réels expliqueront ensuite les partitions de calcul, les traitements, les étapes et les tâches.
