Block 2 · Exercise 1
Open the notebook 13-spark-architecture-starter in your workspace and run the setup cell, which starts the Spark session. Most steps end in a question whose answer is on your own screen, in the Spark UI, so open the UI in a second browser tab and keep it there.
Do one part at a time, 1a, 1b and so on, and compare with the expected output where there is one. The solution is in 13-spark-architecture-solution, under every part on the exercise site, and at the back of the exercises PDF.
Find the Spark UI for your session. In a Fabric notebook it sits under the session indicator at the bottom of the notebook, and in the Monitor hub under your running application. Open it in a second browser tab.
The slide Reading the Spark UI lists the tabs and what each one shows.
Session indicator at the bottom of the notebook, then the link to the Spark UI. The Monitor hub lists the same application under Monitor.
Open the Executors tab. Write down how many executors you have, how many cores each has, and how much memory is listed against each. Run the cell of Step 1, which prints what the session reports about itself, and compare.
The Executors tab has a row for the driver as well. Count the executors separately from it.
executor entries : <a small whole number>
hosts and memory : <a map with one entry per executor and the driver>
cores for tasks : <a whole number>The numbers depend on the pool your session runs on.
status = spark.sparkContext._jsc.sc().getExecutorMemoryStatus()
print("executor entries :", status.size())
print("hosts and memory :", status)
print("cores for tasks :", spark.sparkContext.defaultParallelism)A local Spark reports one executor entry, the driver. A Fabric session on the course pool reports the driver plus the pool's executors. Keep the three numbers : they explain every task count you see later.
Run the cell that counts the flights. Go to the Jobs tab. One count() started two jobs. How many stages does the older of the two have?
The slide Single Stage Job shows what a job without a shuffle looks like.
One stage : the tasks that read the files and count each partition.
Click into the older job, then into its stage. How many tasks are in that stage? Does the number match the number of files, the number of cores, or neither? Then print the partition count of flights and compare.
A task processes one partition. flights.rdd.getNumPartitions() gives the count.
partitions : <a whole number>
rows : <the number of flights>The partition count equals the task count of the stage. Both depend on the data files and the session.
print("partitions :", flights.rdd.getNumPartitions())
print("rows :", flights.count())The task count is the partition count of the parquet files. Nothing you wrote said how many partitions to use, so Spark chose when it read them.
Open the newer job. Which of its stages is marked skipped, and how many tasks ran in the other one?
The slide Skipped Stages and Caching says when Spark lists a stage as skipped.
A count on a DataFrame counts each partition, then sends the partial counts through a shuffle to one task that adds them up. With adaptive query execution on, which is the default, Spark runs the work before a shuffle as a job of its own, measures what it wrote, and then plans the rest. The later job lists the stage that already ran as skipped. The stage that ran in the later job has one task, the one that adds up the partial counts. The job with one stage is the first of the two.
Run the cell that groups the flights by carrier and shows the counts. Go to the Jobs tab. The cell started two jobs again, the same shape as in Step 2. Open the newer job. How many stages does it have, and which one is skipped?
Compare with what you saw for the newer job in 2c.
Two stages. The stage before the shuffle already ran in the older job and is listed as skipped ; the stage after it ran.
Which operation caused the shuffle this time, and which one caused it in Step 2? Print the plan of by_carrier and find the shuffle in it.
by_carrier.explain() prints the plan. A shuffle shows up as a node named Exchange.
== Physical Plan ==
AdaptiveSparkPlan ...
+- HashAggregate(keys=[carrier...], functions=[count(1)])
+- Exchange hashpartitioning(carrier...)
+- HashAggregate(keys=[carrier...], functions=[partial_count(1)])
+- FileScan parquet ...The exact text and numbers vary. The shape is an aggregate, an Exchange, and an aggregate again.
by_carrier.explain()groupBy adds the shuffle, because every row for one carrier has to meet on one executor. The Exchange node in the plan is that shuffle. In Step 2 the count itself needed one.
Open the Stages tab and find the stage with the shuffle. How much data was written in Shuffle Write, and how much was read on the other side? Compare the shuffle write with the size of the data the job read.
The shuffle write is far smaller than the data the job read. Spark aggregates within each partition before the shuffle, so only the partial counts cross the network.
Stay on the stage you opened in 3c. Look at the task table at the bottom, or at the summary metrics above it. What was the duration of the Max task, and of the Median task? How far apart are they? Write both numbers down : Step 6 compares against them.
The summary metrics table has a row per statistic, with Min, 25th percentile, Median, 75th percentile and Max as columns.
A large gap between median and max in one stage means the work was not spread evenly : the data is skewed.
Count the flights per origin airport and show the ten busiest.
groupBy("origin").count(), then sort on the count, descending.
+------+------+
|origin| count|
+------+------+
| ...| ...|
...
+------+------+
only showing top 10 rowsTen airports, the count falling down the column. Which airports and what counts depend on the dataset size.
(flights.groupBy("origin").count().orderBy(F.desc("count"))).show(10)The row counts per airport are what the gap between median and max reflects : a few hub airports carry far more flights than the rest.
Run the cell that caches by_carrier and then calls count() on it twice. Look at the jobs this cell started in the Jobs tab. How many jobs did the first count() start, and how many did the second?
The jobs of one cell sit together in the Jobs tab, the newest at the top.
On Fabric the first count ran as four jobs and the second as two. Your numbers may differ with the session, so compare the first against the second : the second starts fewer.
Open the first job of the second count. Which stage is marked skipped, and how many tasks ran in the stage after it?
Two things cause skipped stages here. One is the same as in 2c. Read the slide Skipped Stages and Caching for the other.
Skipped stages appear in both counts, for two reasons. Inside one action, every job after the first lists the stages already run, as in Step 2. In the second count, the stage that reads the flights is skipped as well, because the cache already holds what it produced ; the stage after it reads the cached partitions instead.
A skipped stage is Spark reusing what it already has : shuffle output from an earlier job, or a cached copy. Without the cache, a second count() reads the flights again, because count() builds a new query each time.
Open the Storage tab. What is listed there, how many partitions does it hold, and what fraction is cached in memory?
The Storage tab lists the cached by_carrier result, with its partitions and the fraction held in memory.
Empty the cache again and watch the Storage tab empty.
The opposite of cache() is a method on the same DataFrame.
cache releasedby_carrier.unpersist()
print("cache released")Cache by_carrier once more and time two count() calls with time.perf_counter. Print one line per call.
Time each call the way you would time any Python function : take time.perf_counter() before and after.
first action <seconds>s
second action <seconds>sThe second count is faster than the first. The times themselves vary from run to run.
import time
by_carrier.cache()
for label in ("first action", "second action"):
start = time.perf_counter()
by_carrier.count()
print(f"{label:16} {time.perf_counter() - start:6.2f}s")Run the cell with the join of flights and airports, the largest thing you run today. While it runs, watch the Jobs tab. How many stages did the whole thing take?
+------------+---------+-------+
|municipality|avg_delay|flights|
+------------+---------+-------+
| ...| ...| ...|
...
+------------+---------+-------+Up to ten cities, avg_delay falling down the column, every flights above 10000. The cities and numbers depend on the dataset size.
worst = (flights
.join(airports, flights.origin == airports.iata_code)
.groupBy("municipality")
.agg(F.avg("dep_delay").alias("avg_delay"),
F.count("*").alias("flights"))
.filter(F.col("flights") > 10000)
.orderBy(F.desc("avg_delay")))
worst.show(10)Which stage took the longest, and what was it doing? Compare the Max and Median task time in that stage against the two numbers from 4a.
Open the Stages tab and sort on the Duration column.
Read the stage's name and its input and shuffle columns to see what it was doing. If its Max task time is far above its median, the work was uneven, as in Step 4.
Open the SQL tab, click the query, and look at the plan diagram. Can you find the join in it? Does the plan say BroadcastHashJoin or SortMergeJoin? Is there a node in the plan you did not write? Print the plan in the notebook as well.
worst.explain() prints the same plan as text. Look for Exchange nodes : they mark the stage boundaries.
== Physical Plan ==
AdaptiveSparkPlan ...
+- Sort ...
+- Exchange rangepartitioning(avg_delay DESC ...)
+- Filter (flights > 10000)
+- HashAggregate(keys=[municipality...])
+- Exchange hashpartitioning(municipality...)
+- HashAggregate(keys=[municipality...])
+- BroadcastHashJoin ...The shape is a scan and a join at the bottom, two aggregates with an Exchange between them, then the filter and the sort. Names and numbers vary.
worst.explain()The airports table is small, so Spark may have chosen to broadcast it instead of shuffling both sides. Look for BroadcastHashJoin or SortMergeJoin, and for the Exchange nodes that mark the stage boundaries. On Fabric the SQL tab may also carry entries the platform itself created ; yours is the one whose time matches the job you just ran.
Click the session indicator and stop the session.
Nothing to run. A session holds its executors until the idle timeout, which is 60 minutes on the course pool, whether you are running anything or not. With the whole class on one capacity, a forgotten session is the reason somebody else cannot start theirs.
spark.conf.set("spark.sql.adaptive.enabled", False) and compare the stage count and the plan against the adaptive run. What did AQE change? Set it back to True afterwards.flights.repartition(4) and rerun Step 2. What happens to the task count, and what happens to the time?