Revenue per country, billions of rows
A shop wants one number each Monday: revenue per country. The orders table has three billion rows spread over twenty machines.
You do not want to write twenty programs. You want to describe the answer once and let the system split the work.
By the end of this lesson
Note
I can express a data job as a Spark pipeline.
Key words
| Word | What it means | Example |
|---|---|---|
| DataFrame | a table of rows and named columns that Spark works on | |
| Transformation | a step that makes a new table from one | filter |
| Action | a step that runs the plan and gives an answer | count |
| Lazy | waiting to run until an answer is asked for | |
| Partition | one slice of a table, held by one worker | |
| Job | the work Spark does for one action |
The cells students fill in are left blank in this preview.
A DataFrame is a table you can query
MapReduce makes you write the map and reduce steps. Spark lets you describe the result on a DataFrame: a table with named columns.
You can ask a DataFrame questions with code, such as filter and groupBy, or with SQL. Both turn into the same plan. The lab has a file of 24 orders loaded as orders.
Look at three rows
curl -s -d 'SELECT * FROM orders LIMIT 3' http://spark:8000/sqlShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Count the different products
curl -s -d 'orders.select("product").distinct().count()' http://spark:8000/dfShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Transformations wait, actions run
A transformation such as filter or groupBy makes a new DataFrame, but reads nothing yet. Spark only adds the step to its plan. This is being lazy.
An action such as count or collect asks for an answer. Spark then runs the whole plan as a job.
Worked example: steps and their kind
| Step | Kind | Does it read data? |
|---|---|---|
| filter(orders.quantity > 3) | Transformation | No, it adds to the plan |
| select("country") | Transformation | No |
| groupBy("country").sum("quantity") | Transformation | No |
| count() | Action | Yes, it runs the plan |
| collect() | Action | Yes, and brings every row back |
The cells students fill in are left blank in this preview.
A plan that runs on demand
# A plan is a list of steps. Adding a step reads no data.
rows = [["UK", 2], ["Japan", 1], ["UK", 5]]
plan = []
plan.append(["keep_more_than", 1])
print("Plan:", plan)
def collect(rows, plan):
# An action runs the whole plan over the rows.
for step in plan:
if step[0] == "keep_more_than":
kept = []
for row in rows:
if row[1] > step[1]:
kept.append(row)
rows = kept
return rows
print("Result:", collect(rows, plan))
# Change the 1 in the plan to 2 and run again.
Run a plan: count the big orders
curl -s -d 'orders.filter(orders.quantity > 3).count()' http://spark:8000/dfShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
A common mistake
Warning
Many people think a filter line reads the data straight away. It does not: it only adds a step to the plan. Spark checks at once that the columns exist. A bad value inside the data shows up only when an action runs the plan.
Partitions let workers share the work
Spark cuts a DataFrame into partitions. Each worker sums its own partition. Spark then adds the partial sums.
This is the map and reduce idea again, without you writing it. The Spark UI shows each job and its plan, so you can see what ran.
Partial sums, then a total
# Three partitions of the quantity column, each held by a different worker.
partitions = [[2, 1, 3], [4, 2], [5, 1]]
partial = []
for part in partitions:
total = 0
for quantity in part:
total = total + quantity
partial.append(total)
print("Partial sums:", partial)
grand = 0
for value in partial:
grand = grand + value
print("Total:", grand)
# Move the 5 into the first partition and run again.
Ask how many partitions orders has
curl -s -d 'orders.rdd.getNumPartitions()' http://spark:8000/dfShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Cut orders into three, count each
curl -s -d 'orders.repartition(3).rdd.glom().map(len).collect()' http://spark:8000/dfShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Units sold for each product
curl -s -d 'SELECT product, sum(quantity) AS units FROM orders GROUP BY product ORDER BY units DESC, product' http://spark:8000/sqlShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
The Spark UI shows each job
Open Spark UI from the lab bar. The Jobs tab lists the work done for each action. The SQL / DataFrame tab lists each query with its plan.
Run a Run block above again and watch a new entry appear. This is how an engineer finds out why a pipeline is slow.
Going further: an average per product
Write one SQL query of your own for the /sql Run. For each product it gives the number of orders and the average quantity per order, largest average first.
Before you run it, predict which product comes first. Then open the Spark UI and find the job your query made.