SpiralTrain
Exercises › Block 1 · Exercise 12

Block 1 · Exercise 12

PySpark Operation

Starter notebook
12-pyspark-operation-starter
Fabric path
/lakehouse/default/Files/data/solutions/12.pyspark-operation/

Open the notebook 12-pyspark-operation-starter in your workspace and run its first cell. It takes the session Fabric started. Each step has a cell with a comment where your code goes, and the data is six made-up rows.

Do one part at a time, 1a, 1b and so on, and compare with the expected output before you go on. The solution is in 12-pyspark-operation-solution, under every part on the exercise site, and at the back of the exercises PDF.

Step 1DataFrame

1a

The starter holds the list data and the list columns. Make a DataFrame df from them with spark.createDataFrame, and give columns as the schema. Print df. Look at the Spark jobs panel under the cell.

Hint 1

createDataFrame takes the data first. The column names go in its schema argument.

Check your output
DataFrame[Name: string, Age: bigint, Department: string]

The Spark jobs panel stays empty.

Show solutionHide solution
python
df = spark.createDataFrame(data, schema=columns)
print(df)

Step 2Show

2a

Show df. Look under the cell : does a job appear this time?

Hint 1

The slide PySpark Step by Step says what makes Spark run.

Check your output
+-----+---+----------+
| Name|Age|Department|
+-----+---+----------+
|Alice| 34|     Sales|
|  Bob| 45| Marketing|
|Cathy| 29|        IT|
|David| 40|        HR|
|  Eve| 38|     Sales|
|Frank| 30| Marketing|
+-----+---+----------+
Show solutionHide solution
python
df.show()

Step 3Operations

3a

Make df_filtered with the people older than 30. The starter imports col.

Hint 1

df.filter takes a condition. col("Age") > 30 is a condition.

Show solutionHide solution
python
df_filtered = df.filter(col("Age") > 30)

3b

Make df_selected with the columns Name and Department.

Show solutionHide solution
python
df_selected = df.select("Name", "Department")

3c

Make df_grouped with the average Age per Department. Name the average column Average_Age. Print the column names of df_grouped, and run the cell. How long did it take, and how many jobs appeared?

Hint 1

groupBy("Department") is followed by an aggregation. The column it makes is called avg(Age).

Hint 2

withColumnRenamed takes the old name and the new name.

Check your output
['Department', 'Average_Age']

No job appears, and the cell is done in a moment.

Show solutionHide solution
python
df_grouped = df.groupBy("Department").avg("Age").withColumnRenamed("avg(Age)", "Average_Age")
print(df_grouped.columns)

3d

Show df_filtered, df_selected and df_grouped, each with a line of text before it. Count the jobs.

Hint 1

The slide PySpark Step by Step says when Spark runs the work of a transformation.

Check your output
Filtered DataFrame (Age > 30):
+-----+---+----------+
| Name|Age|Department|
+-----+---+----------+
|Alice| 34|     Sales|
|  Bob| 45| Marketing|
|David| 40|        HR|
|  Eve| 38|     Sales|
+-----+---+----------+

Selected Columns (Name and Department):
+-----+----------+
| Name|Department|
+-----+----------+
|Alice|     Sales|
|  Bob| Marketing|
|Cathy|        IT|
|David|        HR|
|  Eve|     Sales|
|Frank| Marketing|
+-----+----------+

Grouped by Department with Average Age:
+----------+-----------+
|Department|Average_Age|
+----------+-----------+
|     Sales|       36.0|
| Marketing|       37.5|
|        IT|       29.0|
|        HR|       40.0|
+----------+-----------+

The rows of the grouped table can come in another order.

Show solutionHide solution
python
print("Filtered DataFrame (Age > 30):")
df_filtered.show()

print("Selected Columns (Name and Department):")
df_selected.show()

print("Grouped by Department with Average Age:")
df_grouped.show()

3e

Print the number of rows of df_filtered. Write a comment that names the people who are not in it and says why.

Hint 1

A DataFrame has a count() method, and it is an action.

Hint 2

Compare the ages of the missing people with the condition of 3a.

Check your output
4
Show solutionHide solution
python
print(df_filtered.count())
# Cathy is 29 and Frank is 30. The condition is Age > 30, so an age of exactly 30 is left out.

Step 4Plan

4a

Print the plan of df_grouped with explain(). Find the line with Exchange.

Check your output
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[Department#2], functions=[avg(Age#1L)])
   +- Exchange hashpartitioning(Department#2, 200), ENSURE_REQUIREMENTS, [plan_id=97]
      +- HashAggregate(keys=[Department#2], functions=[partial_avg(Age#1L)])
         +- Project [Age#1L, Department#2]
            +- Scan ExistingRDD[Name#0,Age#1L,Department#2]

The plan has this shape. The numbers after the # and the plan_id vary, and so does the last line, which depends on where the data came from.

Show solutionHide solution
python
df_grouped.explain()

4b

Read the plan from the bottom up. Write a comment that says what each level does, and what the Exchange is.

Hint 1

The slide PySpark Architecture shows that the workers are other machines. What has to happen to rows that share a Department but sit on different machines?

Show solutionHide solution
python
# Bottom up : the rows come in, they are grouped and partly averaged on each machine, they are
# exchanged between machines so that every Department ends up in one place, and they are averaged
# again. The Exchange is a shuffle : the most expensive thing in Spark, and the subject of day 4.

Step 5Collect

5a

Collect df_grouped into a Python variable collected_data and print it.

Hint 1

collect() is an action.

Check your output
Collected Data: [Row(Department='Sales', Average_Age=36.0), Row(Department='Marketing', Average_Age=37.5), Row(Department='IT', Average_Age=29.0), Row(Department='HR', Average_Age=40.0)]

The rows can come in another order.

Show solutionHide solution
python
collected_data = df_grouped.collect()
print("Collected Data:", collected_data)

5b

Print the type of the first element of collected_data.

Check your output
<class 'pyspark.sql.types.Row'>
Show solutionHide solution
python
print(type(collected_data[0]))

5c

Write a comment that says where the collected rows are now, why that is a problem on a table with millions of rows, and what you would call instead.

Hint 1

The slide Not Enough Memory is about what happens when one process holds more than it can. Which process holds the rows after collect()?

Show solutionHide solution
python
# collect() copies every row out of the cluster into the memory of the driver, the one process of this
# notebook. On six rows that is nothing. On a table of real size it brings the session down.
# Use show(), take(20) or an aggregation instead.

Step 6Where the work happens

6a

List the processes on the driver with ps -eo pid,ppid,args through subprocess.run, and print only the lines that mention java or python, each cut to 120 characters.

Hint 1

subprocess.run(["ps", "-eo", "pid,ppid,args"], capture_output=True, text=True).stdout is the text of the list. Loop over its splitlines().

Check your output
<pid> <ppid> <path>/python ...
<pid> <ppid> <path>/java ... org.apache.spark ...

The list has at least one python line, your notebook, and one java line, the Spark driver. The process numbers, the paths and the number of extra python lines vary.

Show solutionHide solution
python
lines = subprocess.run(["ps", "-eo", "pid,ppid,args"], capture_output=True, text=True).stdout
for line in lines.splitlines():
    if "java" in line or "python" in line:
        print(line[:120])

6b

Write a comment that says which two lines are your notebook and the Spark driver, and why the executors are not in the list.

Hint 1

The slide PySpark Architecture shows where the workers are.

Show solutionHide solution
python
# The python line is the notebook, the java line is the Spark driver. The executors run on other
# machines, so they are not in this list. From here on the Spark UI is the instrument.

6c

Upper-case every Name with a lambda on df.rdd, and print the collected result.

Hint 1

An RDD has a map method. Each element is a Row, and row.Name is its name.

Hint 2

The slide Python Workers and DataFrames says where a Python lambda runs.

Check your output
['ALICE', 'BOB', 'CATHY', 'DAVID', 'EVE', 'FRANK']
Show solutionHide solution
python
print(df.rdd.map(lambda row: row.Name.upper()).collect())

If time permits

  • Replace groupBy("Department").avg("Age") with groupBy("Department").agg(avg("Age"), min("Age"), max("Age")), so the result has three named columns.
  • Run df_grouped.explain(True) and compare the parsed, analysed, optimised and physical plans. Which step resolved the column names?
  • Set spark.conf.set("spark.sql.shuffle.partitions", 4), run Step 3 again, and look at the number of tasks in the Spark jobs panel.
  • Open the Spark UI from the Spark jobs panel and find the SQL tab. Your groupBy is in there as a query, with the plan drawn out.

Tried it yourself first?

The solution is a spoiler. Work through the hints first : a wrong attempt teaches more than a solution you only read.