Introduction
You have written Spark transformations, triggered actions, and saved data. Now you will connect those familiar operations to execution evidence: a plan describing the work and a real event log recording jobs, stages, and tasks. The goal is to explain behavior and read useful errors, rather than memorize internal task counts.
The driver runs your application and coordinates work. In a cluster, executors run tasks on allocated resources; local[2] runs Spark work on this one machine with two execution threads. A job can be divided into stages, and each stage executes tasks over computation partitions. One action can cause more than one job, especially with query planning and adaptive behavior; exact counts can vary.
This independent lab supplies clean orders, customers, two deliberately broken expressions, and a read-only event-summary tool. You should know filtering, grouping, joins, and Parquet outputs. You will not tune resources, collect the general dataset into Python loops, or deploy a cluster. Use the Terminal.
Connect a Familiar Query to Its Execution Plan
In this step, you will inspect a daily regional paid-sales query before requesting its results. The reader, filter, amount, customer join, and aggregation repeat earlier skills.
Here groupBy("order_date", "region") uses two keys together: each distinct date-and-region combination forms one group. East on October 14 is separate from East on October 15 and from West on October 14. Counts and sums are calculated within each combination. This combines the earlier daily and regional questions into one report. explain(mode="formatted") prints the planned operators instead of returning report rows.
The SparkSession also enables a real event log in output/spark-events. Compression is disabled to make that teaching log readable after the application stops. These settings record the execution; they do not substitute fabricated job records. The destination directory is prepared for you.
Start and run the complete inspection script:
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
In Report execution plan, look for the roles of filtering, projection, joining, aggregation, and any Exchange that moves records between computation partitions. Names and plan structure may vary with optimization. An Exchange is an execution boundary, not a new output directory.
Under Report action result, expect October 14 / East / 1 / 20.00, October 14 / West / 1 / 25.00, October 15 / East / 1 / 50.00, and October 15 / West / 1 / 30.00. The total is still 125.00. show and the writer request actual execution of the described work; planning and input metadata handling can also perform preparatory work.
The final display uses spark_partition_id to show computation partition IDs represented by these source records. dropDuplicates lists each observed ID once. IDs and the number observed can vary, and empty partitions need not appear. These runtime groups are different from the date-named directories written by partitionBy in the previous lab. One is work organization; the other is stored data layout.
Read Real Job, Stage, and Task Events
In this step, you will inspect the event log produced by the application you just ran. The provided tool reads JSON event records from that directory and summarizes actual events. You do not need to write a log parser or select a version-specific filename.
Run the summary tool, supplying the event directory as its input argument. The shell's > retains its output for review:
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
Expect an actual application identity, positive Jobs started, Stages submitted, and Tasks ended counts, plus job-to-stage references. The numbers depend on this run and can change with optimization, file splits, and Spark versions. They are observed evidence, not targets to reproduce exactly.
An action requests an answer; Spark may organize the required work into several jobs. A job references stages; stages contain tasks for partitions of work. Shuffle boundaries can separate stages, while several transformations can be combined within one stage. local[2] limits simultaneous execution threads; it does not mean every application has exactly two jobs, stages, or tasks.
For tiny data, JVM startup, scheduling, and file handling can be noticeable relative to the computation itself. A small elapsed-time difference is not a reliable tuning conclusion. Use the plan and event evidence to understand the work first.
Diagnose and Repair an Unknown Column
In this step, you will read a deliberate name-resolution error before correcting its cause. Inspect the prepared broken example:
cat /home/labex/project/tools/broken-column.py
The schema declares quantity, while its final select asks for quanity. Save the actual failure output. > redirects ordinary output to the log file; 2>&1 sends standard error to that same destination, so the traceback is saved too. The next command is expected to fail.
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
Find the unresolved column name and any suggested valid names. A long traceback contains implementation details, but the column reference points to the correction. The input does not need reinstallation or conversion into a Python loop.
Create a corrected program with the declared column name, leaving the broken training fixture available for comparison:
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
Expect all six IDs with quantities 2, 1, 1, 2, 1, and 2. The projection now resolves against the real schema and saves its actual result. Column names, including spelling, are part of the data contract.
Repair a Filter That Is Not a Boolean Condition
In this step, you will distinguish a valid integer column from a valid filtering condition. Inspect the second prepared example:
cat /home/labex/project/tools/broken-filter.py
filter expects a Boolean expression for each row. col("quantity") alone is an integer, not a true/false test. Capture the deliberate type mismatch. The next command is expected to fail.
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
Look for the filter condition and its non-Boolean integer type. The repair is to express the intended comparison, quantity > 0, using the already learned column operator. Run the corrected complete program:
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
All six clean fixture rows have positive quantities, so all remain. A negative or zero quantity would fail this condition; this particular repair does not apply the separate paid-sales rule.
You now have evidence at three levels: the intended expression and schema, the actual planned/executed work, and correct persisted results after focused repairs. Keep the distinction between stored date directories and computation partitions when explaining execution. Next, you will submit an independent batch script with explicit input and output parameters.
Summary
You connected familiar transformations to a real plan and event log, observed application/job/stage/task evidence without fixing their counts, and distinguished computation partitions from date directories. You repaired an unresolved column and a non-Boolean filter using schema and error evidence. Next, explicit submission parameters will make your batch application independent of a previous process.
