Do you need a cluster for this?
Spark, SQL and plain Python can all answer the same data question, so choosing between them is a real decision. It matters because the wrong choice wastes time, or fails when the data grows.
A team wants revenue for each country. One person reaches for Spark. Another says a ten-line Python loop would do. Which one is right depends on how big the data is. Here you do the same job three ways, then decide with a reason.
By the end of this lesson
Note
I can choose between Spark, SQL and plain Python for one data job.
Key words
| Word | What it means | Example |
|---|---|---|
| DataFrame | a table of rows and named columns that Spark works on | |
| Spark SQL | asking a DataFrame a question in SQL | |
| Shuffle | moving rows between workers so equal keys meet | Exchange |
| Physical plan | the list of steps Spark will really run | |
| In memory | small enough to fit in one machine's memory |
The cells students fill in are left blank in this preview.
One job, three ways
The job is revenue for each country: quantity times price, added up for every country. The lab has 24 orders loaded as orders.
First as a Spark DataFrame expression, then as Spark SQL, then as a plain Python loop over the same file. If the three do not agree, one of them is wrong.
The DataFrame way
curl -s -d 'orders.groupBy("country").agg(F.round(F.sum(orders.quantity * orders.unit_price), 2).alias("revenue")).orderBy("country")' http://spark:8000/dfShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
The SQL way
curl -s -d 'SELECT country, round(sum(quantity * unit_price), 2) AS revenue FROM orders GROUP BY country ORDER BY country' http://spark:8000/sqlShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
The plain Python way
curl -s -d 'import csv, collections; c = collections.Counter(); [c.update({r["country"]: float(r["quantity"]) * float(r["unit_price"])}) for r in csv.DictReader(open("/tmp/lab/orders.csv"))]; sorted(c.items())' http://spark:8000/pyShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
The same job as a loop
# The same job as a plain Python loop: revenue for each country.
orders = [["UK", 2, 25.0], ["Japan", 1, 150.0], ["UK", 1, 150.0], ["Japan", 4, 12.5]]
revenue = {}
for order in orders:
country = order[0]
money = order[1] * order[2]
if country in revenue:
revenue[country] = revenue[country] + money
else:
revenue[country] = money
print(revenue)
# Add ["France", 3, 12.5] to orders and run it again.
A DataFrame is not a list
A DataFrame looks like a table in a loop, but it is a plan. It does not hold the rows in your program, so many list tricks fail.
Ask it for its size with count(), which is an action. Ask for the length with len() and Python refuses.
Ask for its length like a list
curl -s -d 'len(orders)' http://spark:8000/dfShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Ask for its size like Spark
curl -s -d 'orders.count()' http://spark:8000/dfShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
The plan shows where the data moves
Spark can print the physical plan it will run. The word to look for is Exchange. It is a shuffle: rows travel between workers so that equal keys end up together.
A shuffle is the costly part of a big job. Reading a plan tells you where it is.
A physical plan, as Spark prints it
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[country#19], functions=[count(1)])
+- Exchange hashpartitioning(country#19, 4), ENSURE_REQUIREMENTS, [plan_id=244]
+- HashAggregate(keys=[country#19], functions=[partial_count(1)])
+- InMemoryTableScan [country#19]
List the plan's main operators
curl -s -d 'orders.groupBy("country").count()._jdf.queryExecution().simpleString()' http://spark:8000/df | grep -oE "Exchange|HashAggregate|FileScan"Shown 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 Spark is always the better tool because it is the bigger one. For 24 rows it only adds a start-up and a plan. Spark pays off when the data no longer fits on one machine, or the work must be shared.
Worked example: choosing a tool
| The data | A fair choice | Why |
|---|---|---|
| 24 rows in a file | A plain Python loop | Nothing to share out |
| 2 million rows that fit in memory | pandas, or SQL in a database | One machine does it in seconds |
| Billions of rows over many machines | Spark | The work has to be shared |
| A job you will run every night | Whichever is written clearly and checked | Same answer each time |
The cells students fill in are left blank in this preview.
Going further: a different question
Work out the units sold for each product with the DataFrame way, the SQL way and the plain Python way. Do they agree?
Hint: group by product and sum quantity. Then print the plan for your DataFrame line and find its Exchange.