Introduction
Vous avez écrit des transformations Spark, déclenché des actions et enregistré des données. Vous relierez maintenant ces opérations connues à des preuves d’exécution : un plan décrivant le travail et un véritable journal d’événements enregistrant les traitements, étapes et tâches. Le but est d’expliquer le comportement et de lire les erreurs utiles plutôt que de mémoriser des nombres de tâches internes.
Le pilote (driver) exécute votre application et coordonne le travail. Dans un cluster, les exécuteurs exécutent des tâches sur les ressources attribuées ; local[2] exécute le travail Spark sur cette seule machine avec deux fils d’exécution. Un traitement (job) peut être divisé en étapes (stages), et chaque étape exécute des tâches (tasks) sur des partitions de calcul. Une action peut provoquer plusieurs traitements, notamment avec la planification des requêtes et le comportement adaptatif ; les nombres exacts peuvent varier.
Ce lab indépendant fournit des commandes propres, des clients, deux expressions volontairement incorrectes et un outil de synthèse d’événements en lecture seule. Vous devez connaître le filtrage, le regroupement, les jointures et les sorties Parquet. Vous n’ajusterez pas les ressources, ne recueillerez pas tout le jeu de données dans des boucles Python et ne déploierez pas de cluster. Utilisez le Terminal.
Relier une requête connue à son plan d’exécution
Dans cette étape, vous examinerez une requête de ventes payées quotidiennes et régionales avant de demander ses résultats. Le lecteur, le filtre, le montant, la jointure client et l’agrégation reprennent les acquis précédents.
Ici, groupBy("order_date", "region") utilise deux clés ensemble : chaque combinaison distincte de date et de région forme un groupe. East le 14 octobre est distinct de East le 15 octobre et de West le 14 octobre. Les nombres et les sommes sont calculés dans chaque combinaison. Cela combine les questions quotidiennes et régionales précédentes dans un rapport. explain(mode="formatted") affiche les opérateurs planifiés au lieu de renvoyer les lignes du rapport.
La SparkSession active aussi un véritable journal d’événements dans output/spark-events. La compression est désactivée pour rendre ce journal pédagogique lisible après l’arrêt de l’application. Ces paramètres enregistrent l’exécution ; ils ne remplacent pas les événements réels par des enregistrements de traitements fabriqués. Le répertoire de destination est préparé pour vous.
Commencez et exécutez le script d’examen complet :
cd /home/labex/project
cat > /home/labex/project/inspect_execution.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, sum as spark_sum, spark_partition_id
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = (
SparkSession.builder.appName("InspectRetailExecution").master("local[2]")
.config("spark.eventLog.enabled", "true")
.config("spark.eventLog.dir", "file:///home/labex/project/output/spark-events")
.config("spark.eventLog.compress", "false")
.config("spark.eventLog.rolling.enabled", "false")
.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")
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
matched = paid.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(
col("o.order_date").alias("order_date"), col("c.region").alias("region"), col("o.order_amount").alias("order_amount")
)
report = matched.groupBy("order_date", "region").agg(
count("*").alias("paid_order_count"), spark_sum("order_amount").alias("revenue")
)
print("Report execution plan:")
report.explain(mode="formatted")
print("Report action result:")
report.orderBy("order_date", "region").show(truncate=False)
report.write.mode("overwrite").parquet("/home/labex/project/output/daily-regional-sales")
print("Computation partition IDs represented by source rows:")
orders.select(spark_partition_id().alias("compute_partition")).dropDuplicates().orderBy("compute_partition").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_execution.py > /home/labex/project/output/execution-plan-and-result.txt
cat /home/labex/project/output/execution-plan-and-result.txt
Dans Report execution plan, repérez les rôles du filtrage, de la projection, de la jointure, de l’agrégation et de tout Exchange déplaçant des enregistrements entre partitions de calcul. Les noms et la structure du plan peuvent varier avec l’optimisation. Un Exchange est une frontière d’exécution, et non un nouveau répertoire de sortie.
Sous Report action result, vous devez obtenir 14 octobre / East / 1 / 20.00, 14 octobre / West / 1 / 25.00, 15 octobre / East / 1 / 50.00 et 15 octobre / West / 1 / 30.00. Le total reste 125.00. show et l’écriture demandent une véritable exécution du travail décrit ; la planification et la gestion des métadonnées d’entrée peuvent aussi effectuer un travail préparatoire.
L’affichage final utilise spark_partition_id pour montrer les identifiants des partitions de calcul représentées par ces enregistrements sources. dropDuplicates liste chaque identifiant observé une fois. Les identifiants et leur nombre peuvent varier, et les partitions vides peuvent ne pas apparaître. Ces groupes d’exécution diffèrent des répertoires nommés par date écrits par partitionBy dans le lab précédent. L’un est une organisation du travail ; l’autre une organisation des données stockées.
Lire les événements réels de traitements, étapes et tâches
Dans cette étape, vous examinerez le journal d’événements produit par l’application que vous venez d’exécuter. L’outil fourni lit les enregistrements JSON d’événements dans ce répertoire et résume les événements réels. Vous n’avez pas besoin d’écrire un analyseur de journaux ni de sélectionner un nom de fichier propre à une version.
Exécutez l’outil de synthèse en fournissant le répertoire d’événements comme argument d’entrée. Le > du shell conserve sa sortie pour révision :
bigdata-python /home/labex/project/tools/summarize-events.py /home/labex/project/output/spark-events > /home/labex/project/output/event-summary.txt
cat /home/labex/project/output/event-summary.txt
Vous devez trouver une véritable identité d’application, des nombres positifs pour Jobs started, Stages submitted et Tasks ended, ainsi que des références des traitements vers les étapes. Les nombres dépendent de cette exécution et peuvent changer avec l’optimisation, les découpages de fichiers et les versions Spark. Ce sont des preuves observées, et non des cibles à reproduire exactement.
Une action demande une réponse ; Spark peut organiser le travail requis en plusieurs traitements. Un traitement référence des étapes ; les étapes contiennent des tâches pour des partitions de travail. Les frontières de shuffle peuvent séparer les étapes, tandis que plusieurs transformations peuvent être combinées dans une étape. local[2] limite les fils d’exécution simultanés ; cela ne signifie pas que chaque application possède exactement deux traitements, étapes ou tâches.
Pour de petites données, le démarrage de la JVM, la planification et la gestion des fichiers peuvent être perceptibles par rapport au calcul lui-même. Une petite différence de temps écoulé n’est pas une conclusion d’optimisation fiable. Utilisez d’abord le plan et les preuves d’événements pour comprendre le travail.
Diagnostiquer et corriger une colonne inconnue
Dans cette étape, vous lirez une erreur volontaire de résolution de nom avant de corriger sa cause. Examinez l’exemple incorrect préparé :
cat /home/labex/project/tools/broken-column.py
Le schéma déclare quantity, tandis que son select final demande quanity. Enregistrez la véritable sortie de l’échec. > redirige la sortie ordinaire vers le journal ; 2>&1 envoie l’erreur standard à cette même destination pour enregistrer aussi la trace d’erreur. La commande suivante doit échouer.
bigdata-python /home/labex/project/tools/broken-column.py > /home/labex/project/output/column-error.log 2>&1
tail -n 20 /home/labex/project/output/column-error.log
Trouvez le nom de colonne non résolu et les éventuelles suggestions de noms valides. Une longue trace contient des détails de mise en œuvre, mais la référence de colonne indique la correction. L’entrée ne nécessite ni réinstallation ni conversion en boucle Python.
Créez un programme corrigé avec le nom de colonne déclaré, en laissant l’exemple incorrect disponible pour comparaison :
cat > /home/labex/project/fixed_column.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("OrderExpressionExample").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")
fixed = orders.select("order_id", "quantity")
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-quantities")
spark.stop()
PYTHON
bigdata-python /home/labex/project/fixed_column.py > /home/labex/project/output/fixed-column-result.txt
cat /home/labex/project/output/fixed-column-result.txt
Vous devez retrouver les six identifiants avec les quantités 2, 1, 1, 2, 1 et 2. La projection se résout maintenant avec le véritable schéma et enregistre son résultat réel. Les noms de colonnes, y compris leur orthographe, font partie du contrat de données.
Corriger un filtre qui n’est pas une condition booléenne
Dans cette étape, vous distinguerez une colonne entière valide d’une condition de filtrage valide. Examinez le second exemple préparé :
cat /home/labex/project/tools/broken-filter.py
filter attend une expression booléenne pour chaque ligne. col("quantity") seul est un entier, et non un test vrai/faux. Enregistrez l’incompatibilité volontaire de type. La commande suivante doit échouer.
bigdata-python /home/labex/project/tools/broken-filter.py > /home/labex/project/output/filter-error.log 2>&1
tail -n 20 /home/labex/project/output/filter-error.log
Repérez la condition de filtre et son type entier non booléen. La réparation consiste à exprimer la comparaison prévue, quantity > 0, avec l’opérateur de colonne déjà appris. Exécutez le programme complet corrigé :
cat > /home/labex/project/fixed_filter.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("OrderExpressionExample").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")
fixed = orders.filter(col("quantity") > 0)
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-orders")
spark.stop()
PYTHON
bigdata-python /home/labex/project/fixed_filter.py > /home/labex/project/output/fixed-filter-result.txt
cat /home/labex/project/output/fixed-filter-result.txt
Les six lignes propres de l’exercice ont des quantités positives ; elles restent donc toutes. Une quantité négative ou nulle échouerait à cette condition ; cette réparation particulière n’applique pas la règle distincte des ventes payées.
Vous disposez maintenant de preuves à trois niveaux : l’expression et le schéma prévus, le travail réellement planifié et exécuté, puis les résultats persistés corrects après des réparations ciblées. Conservez la distinction entre répertoires de dates stockés et partitions de calcul lorsque vous expliquez l’exécution. Vous soumettrez ensuite un script par lots indépendant avec des paramètres d’entrée et de sortie explicites.
Résumé
Vous avez relié les transformations connues à un véritable plan et à un journal d’événements, observé les preuves d’application, de traitements, d’étapes et de tâches sans en figer les nombres, puis distingué partitions de calcul et répertoires de dates. Vous avez corrigé une colonne non résolue et un filtre non booléen avec les preuves du schéma et des erreurs. Des paramètres de soumission explicites rendront ensuite votre application par lots indépendante d’un processus précédent.
