Introduction
Une commande connaît ses identifiants client et produit, tandis que des tables de référence distinctes décrivent le lieu de résidence du client et la catégorie du produit. Une jointure relie les enregistrements par ces clés pour qu’un rapport réponde à des questions comme le revenu par région ou catégorie de produit.
Vous enrichirez les commandes payées une référence à la fois, sélectionnerez des colonnes de sortie sans ambiguïté et rapprocherez les totaux. Vous comparerez aussi INNER JOIN et LEFT JOIN lorsqu’un client manque. Les clés clients et produits fournies sont uniques ; chaque commande correspondante doit trouver un enregistrement dans chaque référence. Des clés de référence dupliquées pourraient multiplier les lignes jointes et gonfler le revenu ; la relation prévue compte donc.
Ce lab indépendant prépare des commandes propres, deux clients, trois produits et un exemple distinct sans correspondance. Vous devez connaître les lectures typées, les filtres, les montants décimaux, l’agrégation et la sortie Parquet minimale. Aucun Hive, SQL, Hadoop, pandas ou configuration de cluster n’est nécessaire. Utilisez le Terminal.
Ajouter les régions des clients avec une clé explicite
Dans cette étape, vous joindrez les quatre commandes payées à leurs enregistrements clients uniques. Examinez la relation source :
cd /home/labex/project
cat /home/labex/project/data/orders.csv
cat /home/labex/project/data/customers.csv
Plusieurs commandes peuvent partager un customer_id, tandis que la référence contient une ligne pour c1 et une pour c2. Des identifiants égaux relient les enregistrements ; associer les positions de lignes ne serait pas une règle valide. c1 fournit East et c2 fournit West.
La méthode join du DataFrame accepte l’autre DataFrame, une condition de correspondance et un type de jointure. inner conserve uniquement les enregistrements correspondants. Les alias de DataFrame alias("o") et alias("c") donnent aux entrées des noms de référence courts. col("o.customer_id") identifie la colonne côté commandes ; col("c.customer_id") celle côté référence.
Les deux entrées contiennent une colonne customer_id. Le select explicite choisit la clé côté commandes et chaque champ voulu, évitant les noms dupliqués ambigus. Un alias de DataFrame nomme une référence d’entrée ; un alias de colonne nomme un champ de résultat. Enregistrez et exécutez cette version complète, dont le lecteur et le calcul de montant reprennent les acquis précédents :
cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("EnrichOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")
with_customers = (
paid.alias("o")
.join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
.select(
col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
col("c.region").alias("region"),
)
)
with_customers.orderBy("order_id").show(truncate=False)
with_customers.write.mode("overwrite").parquet("/home/labex/project/output/orders-with-regions")
spark.stop()
PYTHON
bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/customer-join.txt
cat /home/labex/project/output/customer-join.txt
Vous devez obtenir quatre lignes payées enrichies. o1 et o4 doivent être dans East, o2 et o6 dans West ; leurs montants restent 20.00, 25.00, 50.00 et 30.00. Quatre entrées payées avec des correspondances uniques complètes produisent quatre lignes jointes. La sortie Parquet préserve les champs typés et ajoute region sans modifier le CSV brut.
Ajouter les catégories de produits et rapprocher le revenu
Dans cette étape, vous ajouterez une seconde référence avec product_id. Examinez la correspondance des produits :
cat /home/labex/project/data/products.csv
p1 et p2 appartiennent à Office (bureau) ; p3 appartient à Home (maison). Une jointure par customer_id serait la mauvaise relation pour cette table. Conservez les champs intermédiaires des commandes enrichies et sélectionnez uniquement la catégorie depuis la référence produits.
Cette version complète conserve la première jointure et ajoute la seconde. La projection explicite conserve un seul product_id plutôt que deux clés de même nom. Ensuite, groupBy et l’agrégat sum déjà connus calculent le revenu par catégorie :
cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("EnrichOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")
with_customers = (
paid.alias("o")
.join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
.select(
col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
col("c.region").alias("region"),
)
)
enriched = (
with_customers.alias("e")
.join(products.alias("p"), col("e.product_id") == col("p.product_id"), "inner")
.select(
col("e.order_id").alias("order_id"), col("e.customer_id").alias("customer_id"),
col("e.product_id").alias("product_id"), col("e.quantity").alias("quantity"),
col("e.unit_price").alias("unit_price"), col("e.order_status").alias("order_status"),
col("e.order_date").alias("order_date"), col("e.order_amount").alias("order_amount"),
col("e.region").alias("region"), col("p.category").alias("category"),
)
)
category_sales = enriched.groupBy("category").agg(spark_sum("order_amount").alias("revenue"))
enriched.write.mode("overwrite").parquet("/home/labex/project/output/enriched-orders")
category_sales.write.mode("overwrite").parquet("/home/labex/project/output/category-sales")
enriched.orderBy("order_id").show(truncate=False)
category_sales.orderBy("category").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/product-join.txt
cat /home/labex/project/output/product-join.txt
Le résultat enrichi doit toujours contenir quatre commandes payées, maintenant avec région et catégorie. Le rapport par catégorie doit montrer Home 30.00 et Office 95.00. Leur total reste le même revenu payé de 125.00.
Le nombre de lignes restant à quatre et le revenu rapproché fournissent deux contrôles utiles. Aucun ne doit augmenter simplement parce que des colonnes descriptives ont été ajoutées. Les clés de cet exemple sont uniques et complètes ; l’étape suivante explore une correspondance manquante séparément plutôt que de modifier les totaux de ventes principaux.
Comparer INNER JOIN et LEFT JOIN avec un client manquant
Dans cette étape, vous observerez l’effet de chaque type de jointure lorsque la référence ne trouve pas de correspondance. Examinez l’exemple indépendant :
cat /home/labex/project/data/example-orders.csv
demo1 fait référence à c1, tandis que demo2 fait référence à c_missing, absent de customers. Prévoyez d’abord INNER JOIN : seul demo1 doit rester. LEFT JOIN conserve chaque enregistrement d’exemple côté gauche et fournit NULL pour les champs qui proviendraient d’un enregistrement manquant côté droit.
Enregistrez un petit programme séparé pour laisser inchangés le script principal d’enrichissement des ventes et ses données :
cat > /home/labex/project/compare_joins.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("CompareCustomerMatches").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
examples = spark.read.option("header", True).schema("order_id string, customer_id string").csv("/home/labex/project/data/example-orders.csv")
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
inner = examples.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"), col("c.region").alias("region"))
left = examples.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "left").select(col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"), col("c.region").alias("region"))
print("Inner matches:")
inner.orderBy("order_id").show(truncate=False)
print("Left matches:")
left.orderBy("order_id").show(truncate=False)
inner.write.mode("overwrite").parquet("/home/labex/project/output/example-inner")
left.write.mode("overwrite").parquet("/home/labex/project/output/example-left")
spark.stop()
PYTHON
bigdata-python /home/labex/project/compare_joins.py > /home/labex/project/output/join-comparison.txt
cat /home/labex/project/output/join-comparison.txt
Inner matches doit contenir demo1/c1/East. Left matches doit contenir cette ligne plus demo2/c_missing/NULL. Le customer_id côté gauche reste disponible même si la région côté droit manque. NULL est l’absence de région correspondante, et non une région littéralement nommée NULL.
INNER JOIN peut convenir à un rapport dont les correspondances complètes sont déjà établies. LEFT JOIN peut conserver les enregistrements non résolus pour des contrôles de qualité ou une politique métier délibérée. Choisir le type de jointure change les enregistrements conservés ; ce n’est pas une simple préférence d’affichage.
Utiliser les champs enrichis pour une comparaison régionale
Dans cette étape, vous appliquerez le filtrage et le regroupement déjà appris au jeu de données enrichi principal. Le revenu régional toutes catégories doit être East 70.00 et West 55.00. Prévoyez le résultat limité à Office : East reste à 70.00, tandis que West passe à 25.00, car sa commande Home de 30.00 est exclue.
Exécutez le script principal complet avec deux rapports dérivés supplémentaires. Ses jointures restent identiques ; la condition de catégorie change uniquement les lignes de commandes enrichies entrant dans la seconde agrégation :
cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("EnrichOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")
with_customers = (
paid.alias("o")
.join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
.select(
col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
col("c.region").alias("region"),
)
)
enriched = (
with_customers.alias("e")
.join(products.alias("p"), col("e.product_id") == col("p.product_id"), "inner")
.select(
col("e.order_id").alias("order_id"), col("e.customer_id").alias("customer_id"),
col("e.product_id").alias("product_id"), col("e.quantity").alias("quantity"),
col("e.unit_price").alias("unit_price"), col("e.order_status").alias("order_status"),
col("e.order_date").alias("order_date"), col("e.order_amount").alias("order_amount"),
col("e.region").alias("region"), col("p.category").alias("category"),
)
)
category_sales = enriched.groupBy("category").agg(spark_sum("order_amount").alias("revenue"))
enriched.write.mode("overwrite").parquet("/home/labex/project/output/enriched-orders")
category_sales.write.mode("overwrite").parquet("/home/labex/project/output/category-sales")
enriched.orderBy("order_id").show(truncate=False)
category_sales.orderBy("category").show(truncate=False)
regional = enriched.groupBy("region").agg(spark_sum("order_amount").alias("revenue"))
office = enriched.filter(col("category") == "Office").groupBy("region").agg(spark_sum("order_amount").alias("revenue"))
regional.write.mode("overwrite").parquet("/home/labex/project/output/regional-sales")
office.write.mode("overwrite").parquet("/home/labex/project/output/office-regional-sales")
print("All categories by region:")
regional.orderBy("region").show(truncate=False)
print("Office category by region:")
office.orderBy("region").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/regional-comparison.txt
cat /home/labex/project/output/regional-comparison.txt
Vous devez retrouver les résultats régionaux prévus toutes catégories et pour Office seul. Les régions Office totalisent 95.00, correspondant au sous-total de catégorie Office de l’étape précédente. Les régions globales totalisent 125.00.
Ces rapprochements expliquent à la fois la relation de jointure et le sous-ensemble métier sélectionné. Votre jeu de données Parquet enrichi principal contient toujours les quatre commandes payées correctement typées, et la démonstration de correspondance manquante reste séparée. Vous exprimerez ensuite des opérations comparables avec Spark SQL sur une vue DataFrame.
Résumé
Vous avez enrichi les commandes avec des clés clients et produits uniques explicites, sélectionné des colonnes de sortie sans ambiguïté et rapproché les totaux par catégorie et région. INNER JOIN et LEFT JOIN ont produit volontairement des comportements différents en l’absence de correspondance. Vous comparerez ensuite Spark SQL et les expressions DataFrame sur les mêmes données structurées.
