Sélectionner, filtrer et calculer les montants des commandes dans PySpark

Introduction

Une table de commandes correctement typée est un point de départ, pas encore une réponse. Un collègue peut souhaiter quelques colonnes seulement, uniquement les commandes payées et un montant calculé à partir de la quantité et du prix unitaire. Vous exprimerez ces décisions sous forme d’opérations DataFrame Spark tout en gardant les enregistrements originaux disponibles.

Une expression de colonne décrit un calcul ou une condition que Spark appliquera aux lignes. Vous sélectionnerez des champs, combinerez des conditions, ajouterez un montant décimal et donnerez un nom lisible à une colonne de résultat. Ces opérations créent de nouveaux DataFrames ; des actions comme show demandent les résultats réels. Cette distinction aide à expliquer l’exécution paresseuse de Spark sans avoir encore besoin du fonctionnement interne de la planification.

Ce lab fournit indépendamment six commandes de vente propres. Vous devez comprendre le schéma CSV explicite du lab précédent et les bases de Python. Le lecteur complet et la SparkSession sont inclus dans chaque version exécutable ; aucun processus ni fichier précédent n’est donc nécessaire. Utilisez le Terminal et l’environnement bigdata-python préparé.

Sélectionner les colonnes nécessaires à une question

Dans cette étape, vous réduirez les colonnes de la table tout en conservant les six enregistrements de commandes. Commencez par la source :

cd /home/labex/project

cat /home/labex/project/data/orders.csv

select choisit quelles colonnes un nouveau DataFrame expose et dans quel ordre. Il ne filtre pas les lignes. Le script complet ci-dessous reprend le lecteur explicite connu, puis sélectionne l’identifiant de commande, la quantité, le prix et le statut. Les champs de date et de client restent dans df, même si selected ne les expose pas.

Enregistrez le script avec un here-document entre apostrophes et exécutez-le. > écrit le fichier ou remplace la version précédente ; le marqueur PYTHON final termine le texte source :

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

selected = df.select("order_id", "quantity", "unit_price", "order_status")
selected.orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/selected-orders.txt

cat /home/labex/project/output/selected-orders.txt

Vous devez retrouver les six identifiants de commandes avec quatre colonnes affichées. orderBy rend l’affichage prévisible, tandis que show déclenche le travail nécessaire pour renvoyer les lignes. Il s’agit d’une sélection de colonnes, parfois appelée projection. Aucun fichier source n’a changé et aucune commande n’a été retirée.

Combiner des conditions pour sélectionner les lignes

Dans cette étape, vous conserverez les commandes payées ayant une quantité d’au moins deux. col("order_status") désigne une colonne Spark ; la comparer avec == "paid" construit une condition pour Spark, plutôt qu’un unique True ou False Python. >= construit la condition de quantité.

Utilisez & pour combiner les conditions de colonnes Spark selon une logique ET. Entourez chaque comparaison de parenthèses, car la priorité des opérateurs Python compte. Le and ordinaire de Python n’est pas l’opérateur permettant de combiner des objets Column Spark.

Prévoyez les identifiants admissibles : o1, o4 et o6 respectent les deux conditions. o2 est payée, mais possède une quantité d’un. Remplacez et exécutez le script complet :

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

eligible = df.filter((col("order_status") == "paid") & (col("quantity") >= 2))
eligible.select("order_id", "quantity", "unit_price", "order_status").orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/eligible-orders.txt

cat /home/labex/project/output/eligible-orders.txt

Le résultat doit contenir uniquement ces trois identifiants, chacun avec une quantité de 2. filter change les lignes sélectionnées ; select contrôle les colonnes affichées. Vous pouvez combiner les deux opérations tandis que df continue de décrire la source complète à six lignes.

Le seuil de quantité de cette étape est une condition de démonstration, et non une définition générale des ventes. Le prochain calcul inclura les quatre commandes payées, dont o2, car chaque commande payée contribue au revenu.

Calculer et nommer les montants décimaux des commandes

Dans cette étape, vous calculerez le montant de chaque commande payée comme la quantité multipliée par le prix unitaire décimal. withColumn ajoute une colonne nommée à un nouveau DataFrame, ou remplace cette colonne si son nom existe déjà. Ici, order_amount est calculé avec l’expression de multiplication intégrée de Spark.

Le select suivant conserve l’identifiant de commande et donne à la colonne calculée le nom d’affichage amount avec alias. Un alias change le nom de colonne du résultat ; il ne modifie pas l’en-tête CSV. Les quatre commandes payées appartiennent à ce calcul ; les lignes en attente et remboursées sont exclues, sans traiter les remboursements comme des revenus négatifs dans cette règle simplifiée.

Exécutez cette version complète :

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

paid_result = (
    df.filter(col("order_status") == "paid")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/paid-amounts.txt

cat /home/labex/project/output/paid-amounts.txt

Vous devez obtenir o1/20.00, o2/25.00, o4/50.00 et o6/30.00. Des zéros finaux supplémentaires peuvent apparaître. Le schéma doit identifier amount comme un type décimal. Spark peut augmenter la précision décimale pendant la multiplication ; l’essentiel est que le calcul conserve une arithmétique décimale plutôt que de convertir les prix en virgule flottante binaire.

filter, withColumn, select et orderBy décrivent des transformations et renvoient de nouveaux DataFrames. Définir cette chaîne ne matérialise pas immédiatement chaque table intermédiaire. show est une action demandant à Spark d’exécuter le travail requis et d’afficher une réponse. printSchema décrit la structure des colonnes. Vous examinerez plus tard le véritable plan d’exécution derrière une telle chaîne.

Modifier un statut et prévoir son effet

Dans cette étape, vous conserverez le résultat payé et ajouterez une vue d’exercice distincte des commandes en attente. Prévoyez sa sortie avant l’exécution : seule o3 est en attente, avec une quantité d’un et un prix de 10.00 ; son montant est donc 10.00.

Le nouveau pending_result utilise la même chaîne d’expressions en changeant uniquement la valeur du statut. Il part de df, et non du paid_result déjà filtré ; les enregistrements en attente sont donc toujours disponibles :

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

paid_result = (
    df.filter(col("order_status") == "paid")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()

pending_result = (
    df.filter(col("order_status") == "pending")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Pending practice:")
pending_result.orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/status-comparison.txt

cat /home/labex/project/output/status-comparison.txt

Sous Paid orders, les quatre mêmes montants payés doivent rester. Sous Pending practice, vous devez obtenir uniquement o3/10.00. Filtrer la vue payée pour le statut en attente ne renverrait aucune ligne, car ces enregistrements ont déjà été exclus ; partir de la source appropriée compte.

Les deux chaînes sont évaluées par les expressions intégrées de Spark. Vous n’avez pas recueilli l’ensemble du jeu de données dans une liste Python pour le parcourir en boucle. Cela conserve une description du traitement adaptée au moteur d’exécution de Spark. Votre script final conserve le calcul payé et la petite comparaison d’exercice, tandis que le CSV original contient toujours les six commandes.

Résumé

Vous avez sélectionné des colonnes, combiné des conditions de lignes avec des expressions Spark, calculé des montants décimaux de commandes et nommé les colonnes de sortie. La comparaison des vues payées et en attente a montré pourquoi les transformations doivent partir de la source appropriée et comment les actions demandent les résultats. Vous appliquerez ensuite les notions connues de lecture et de sélection aux entrées JSON.