Like all things in technology, Big Data always has yet another thing for you to learn in your quest to continually stay relevant. The topic for today’s learning is the Z-Order strategy used with modern table formats like Apache Iceberg and Delta Lake. Specifically, I’m going to try to show you what it is (at a high-level) and how it compares with the classic sorting that has been used on data lake tables for years.
Before I get started, let me share some other blog posts that I reviewed to help get me up to speed.
Now… my recommendation is to NOT look at these blog posts (well, not yet, at least) if you are just getting started on understanding Z-Order. I would however circle back and review them in the order listed if/when you are wanting a deeper dive into the math behind this approach.
Yep… that graphic is in several of the more detailed blog posts and the reason I felt I should try to explain it more simply that all of that. Of course, if it makes perfect sense already… then what are you doing reading this blog post anyways? 😉
Below is a video that shows what a simple table classically sorted by two fields might look like. This is a strategy that has been used for a long while now and I’m happy to see it become a first-class DDL option in Apache Iceberg. Here’s how to leverage it with Trino’s Iceberg connector.
Sorting works great when you add WHERE clauses to your SQL from left to right If you want to search for something that does not start with the first column in the sort order, it usually isn’t all that much help. It often requires you to scan a lot of data, too.
Applying a Z-Order to the same data with the same sort fields ends up storing the data differently. The video attempts to simplify this for initial understanding. It also shows how this strategy helps when you are searching for something other than using the left-to-right column ordering.
Time for the movie! It’s only 10 minutes long and you know that voice is smooth!
If that went as well as I hope it did, you’ll probably understand these Z’s within Z’s within Z’s diagrams.
And if you do, then my mission is a success!!
If you need more details, loop back up to the blog posts I presented at the beginning of this post. I’m optimistic that for some, this was enough to talk intelligently during your first couple of chats on the subject. For anyone that wants/needs to do some deeper digging, I’m betting it will also help when reviewing the more detailed material.
I’ll leave you with some thoughts about Z-Order as we often have a tendency to use anything we hear about — even if it isn’t the right tool for the job. The blog posts I listed provide you even more “considerations” and “warnings”.
If your table isn’t well into the TB range (maybe even at least 10’s of TBs) then the juice probably won’t be worth the squeeze.
If your prominent querying features filtering from left-to-right along the classicly sorted columns, then just stick with that straightforward approach.
Especially true with streaming data pipelines and frequent batch cycles, you will need to come back and rebuild/compact your data files and Z-Order will be more expensive than classical sorting to produce.
Despite what the Delta Lake folks seem to be saying lately, Z-Order can work great with partitioning on your ludicrously large tables; especially when you are normally querying only a small subset of the existing partitions.
The bird says, “I’m the Python dataframe library with tons of optionality”. The bunny says, “I’m the SQL engine with heaps of optionality”. They instantly became BFFs. The end.
OK.. that “joke” did stink, but using Ibis with Trino for the backend is the complete opposite. It’s optionality2 and that’s a pretty sweet thing!!
This post is a repeat of pystarburst (the dataframe api), but this time using Ibis. As before, this post is NOT attempting to teach you everything you need to know about the DataFrame API, but does provide some insight into the subject matter. Let’s get into it!
I’m using these instructions to continue on with the setup activities and will be using my personal Starburst Galaxy tenant to test with. Unlike in the instructions, I’m just hard-coding my connection details in my boilerplate code (masked, of course).
import os
import ibis
ibis.options.interactive = True
user = "lXXXXXXXXm/aXXXXXXXXXn"
password = "<password>"
host = "lXXXXXXXXXo"
port = "443"
catalog = "tpch"
schema = "tiny"
con = ibis.trino.connect(
user=user, password=password, host=host, port=port,
database=catalog, schema=schema
)
tiny_region = con.table("region")
print(tiny_region[0:5])
Test the boilerplate code
I’m using the CLI, but you could easily run this in your favorite web-based notebook.
$ python3 ibis-main.py
Yep, we basically ran a simple SELECT statement and we can verify in Starburst Galaxy’s Query history screen that it executed.
Explore the API
There’s a lot of solid documentation on the Ibis site, but the Basic operations page serves as good of a starting point as any other to get your hands dirty writing some code.
If interested in comparing side-by-side the DataFrame API code from Ibis with that from PyStarburst, just pull up my pystarburst (the dataframe api) blog alongside this one since I’m porting that code to work with Ibis.
Notice that even though 100 records were requested to be displayed, there are only 7 records that meet this criteria.
Select a second table
Let’s create a DataFrame from the nation table to later join with customer. In the example below, we are chaining methods together instead of assigning each output to a distinct variable as we have done up until now.
# Grab new table, drop 2 cols, and rename 2 others
nationDF = con.table("nation") \
.drop("regionkey", "comment") \
.rename(
dict(
nation_name="name",
n_nationkey="nationkey"
)
)
print(nationDF.head(10))
Join the tables
This is the EXACT same syntax used in PyStarburst (and yes, PySpark, too).
While the creation of multiple DataFrame objects was used above, in practice (as discussed when fetching the nation table) most DataFrame API programmers chain many methods together to look at bit more like this.
The results are the same as before just like in the original PyStarburst post’s code that we just ported to use Ibis. The APIs are different enough that you likely would want to pick one and stick with it instead of trying to use them both daily. For me personally, I’m currently in the PyStarburst camp.
Sure, I work at Starburst which has SOMETHING to do with that, but it is really because it lines up more closely to the PySpark implementation I have spend a number of years working with. If I was starting from scratch I wouldn’t have the prior experience to drive me to think this way and I would likely do more personal research and comparison.
It surely is NOT a dig against Ibis and its optionality of being able to run the same DataFrame program against multiple backend SQL engines. That is an INCREDIBLE foundational feature of this framework. Fortunately, Trino’s connector architecture and long list of integrations offers that same kind of optionality at the SQL engine layer.
Of course, if there is a backend in the list below that you need this flexibility for that is not a supported integration with Trino (quack-quack comes to mind, but isn’t the only one in that list not currently integrated with Trino), then you’ll have to take a hard look at Ibis.
For Trino (and Starburst) fans like myself, it is surely a win if you chose to go down the Ibis path as we do believe in this integration and want to only make it better over time.
As for performance & optimization… I was running all these simple examples on the TPCH connector (and the tiny schema at that) which absolutely does not allow any inference to be made from the limited set of examples I ran for this blog post. One would assume that the CBO would ultimately decide to tackle the problem the same way regardless of which DataFrame API implementation was used.
In fact, we ended up getting a VERY SIMILAR query plan for PyStarburst and Ibis as expected. The DAG on the left is from PyStarburst and the one on the right from the Ibis invocation.
If those pictures look like hieroglyphics and all that CBO & DAG talk was mumbo-jumbo, and you want to learn more, check out these free training modules from Starburst Academy.
Back to the visualizations. Yes, the text is amazingly small (and fuzzy) and is almost completely unreadable, but I do see something on the left that didn’t happen on the right.
The PyStarburst execution ended up running ScanFilterProjects instead of just ScanFilters that Ibis produced. Again… do NOT read anything into this; especially with the data generator connector I was using. It just lets me know I need to do some more side-by-side research.
For that, pushing the TPCH generated data into a modern table format like Iceberg and using a bigger sized schema could then offer some real testing opportunities.
All in all, I’m happy I stayed up until 2:00 am on a work night doing some exploration and sharing my initial findings. If you enjoyed it to, please do let me know by following my blog and/or leaving a comment below. Thanks so much!
After this and that, you might think I’d be done posting PyStarburst DataFrame API examples, but I’m still excited to share a few more. I ended my last PyStarburst post with some examples of Window functions. To help conceptually understand them better, I posted window functions explained. This post will focus on some additional windowing examples with Python and Starburst via the DataFrame API.
Also like in my last post, I’ll share SQL first followed a Python approach. For a dataset, I’ll make it super easy. Starburst Galaxy already has a nice little sample catalog whose demo schema offers up some out-of-this-world tables.
For this post, I’ll explore the astronauts table exclusively.
Note: Take a look at the full code listing at the end of this post to get the boiler plate code you’ll need at the top of your .py file or in your web-based notebook.
-- SQL soln
SELECT *
FROM sample.demo.astronauts
LIMIT 10;
# Python soln
a = session.table("sample.demo.astronauts")
a.show(10);
Both the SQL and Python outputs show this to be a pretty wide table. So wide, I don’t want to show it. We could ask them both to share some more about the columns that make them up.
-- SQL soln
DESC sample.demo.astronauts;
# Python soln
for field in a.schema.fields:
print(field.name +" , "+str(field.datatype))
Note: PySpark has a nice printSchema() function that produces a very nice output. At this early stage in PyStarburst it doesn’t seem it has been implemented, but I’m sure it will surface before long.
And again, BOTH of these produce a BUNCH of text, so I’m just showing the web UI’s output as it is more quickly consumable for most of us.
As I know the table pretty well, let’s just run this highly projected & filtered query to simplify the data we are looking at.
-- SQL soln
SELECT name, nationality,
year_of_mission AS m_yr,
hours_mission AS m_hrs
WHERE name IN ('Nicollier, Claude', 'Ross, Jerry L.')
ORDER BY name, m_yr;
What I wanted to show is that this table is NOT a list of astronauts. It is a table of astronauts’ trips to space. These two fellas are represented multiple times; Jerry went to space 7 times and Claude went 4 times.
Window function examples
Single window for all rows
Let’s see how each mission compares with an OVERALL average across ALL missions. Create a single window that encompasses all input rows and calculates a single average for all output rows.
SELECT name,
year_of_mission AS m_yr,
hours_mission AS m_hrs,
AVG(hours_mission)
OVER() -- the WINDOW is ALL rows
AS avg_all_m_hrs -- NOT typical ^^^^^^^^
FROM sample.demo.astronauts
WHERE name IN ('Nicollier, Claude', 'Ross, Jerry L.')
ORDER BY name, m_yr;
Introduce the over() function without any parameters to create a single window to be used for all input rows.
# trim out the nation column
aDF = twoAs.drop("nationality")
# use an empty parameter over function call
aDF.withColumn("avg_all_m_hrs", F.avg("m_hrs").over()) \
.sort("name", "m_yr").show(20)
Window for each distinct value
Keep all the rows, but calculate an aggregate for each window created for all input rows with the same astronaut.
SELECT name,
year_of_mission AS m_yr,
hours_mission AS m_hrs,
SUM(hours_mission)
-- kinda like a GROUP BY, but you get all rows
OVER (PARTITION BY name)
AS tot_m_hrs
FROM sample.demo.astronauts
WHERE name IN ('Nicollier, Claude', 'Ross, Jerry L.')
ORDER BY name, m_yr;
from pystarburst.window import Window as W
# define the window specification
w2 = W.partition_by("name")
aDF.withColumn("tot_m_hrs", F.sum("m_hrs").over(w2)) \
.sort("name", "m_yr").show(20)
Multiple aggregations on the same window
You can calculate multiple aggregations on the same window specification.
SELECT name,
year_of_mission AS m_yr,
hours_mission AS m_hrs,
AVG(hours_mission)
OVER (PARTITION BY name)
AS avg_m_hrs,
-- we can have more than 1
SUM(hours_mission)
OVER (PARTITION BY name)
AS tot_m_hrs
FROM sample.demo.astronauts
WHERE name IN ('Nicollier, Claude', 'Ross, Jerry L.')
ORDER BY name, m_yr;
Of course, we can also manipulate the values we get back from the window’s aggregate functions, too.
-- determine the percentage of each mission against
-- that astronaut's total
SELECT name,
year_of_mission AS m_yr,
hours_mission AS m_hrs,
ROUND(hours_mission /
SUM(hours_mission) OVER (PARTITION BY name)
* 100.0, -- example: change 0.12 to 120
2) -- round off to decimal places
AS percent_of_tot,
SUM(hours_mission)
OVER (PARTITION BY name)
AS tot_m_hrs
FROM sample.demo.astronauts
WHERE name IN ('Nicollier, Claude', 'Ross, Jerry L.')
ORDER BY name, m_yr;
Using different columns to create multiple windows
For each row you can create additional windows. In the example below, we are adding another one based on all input records that have the same year_of_mission column as the current row.
-- use a second window definition
SELECT name,
year_of_mission AS m_yr,
hours_mission AS m_hrs,
SUM(hours_mission)
OVER (PARTITION BY name)
AS tot_m_hrs,
COUNT()
OVER (PARTITION BY year_of_mission)
AS tot_m_yr_for_all
FROM sample.demo.astronauts
WHERE name IN ('Nicollier, Claude', 'Ross, Jerry L.')
ORDER BY name, m_yr;
Only 1993 had more than one space flight and we see it for both astronauts in our dataset.
Order the window’s contents
By adding an ORDER BY clause within the window definition, we can perform some additional calculations based on position of the current input row in the window. This example shows how you can create a mission number.
SELECT name,
year_of_mission AS m_yr,
hours_mission AS m_hrs,
RANK() OVER (PARTITION BY name
ORDER BY year_of_mission)
AS m_nbr
FROM sample.demo.astronauts
WHERE name IN ('Nicollier, Claude', 'Ross, Jerry L.')
ORDER BY name, m_yr;
In addition to the prior ranking function, you can look forward or backward 1+ record to get values to populate new columns as shown below.
SELECT name,
year_of_mission AS m_yr,
hours_mission AS m_hrs,
LAG(hours_mission, 1)
OVER (PARTITION BY name
ORDER BY year_of_mission)
AS prev_m_hrs,
LEAD(hours_mission, 1)
OVER (PARTITION BY name
ORDER BY year_of_mission)
AS next_m_hrs
FROM sample.demo.astronauts
WHERE name IN ('Nicollier, Claude', 'Ross, Jerry L.')
ORDER BY name, m_yr;
from pystarburst.functions import lag, lead
# lag and lead functions
aDF.withColumn("prev_m_hrs",
lag(aDF.m_hrs, 1).over(w4)) \
.withColumn("next_m_hrs",
lead(aDF.m_hrs, 1).over(w4)) \
.sort("name", "m_yr").show(20)
Create rolling windows
You can bind the window using boundaries such as UNBOUNDED PRECEDING, n PRECEDING, CURRENT ROW, n FOLLOWING, and UNBOUNDED FOLLOWING. The example below has a window that includes the current input row and all previous records based on the sort order.
--calculate running_total_hours
SELECT name,
year_of_mission AS m_yr,
hours_mission AS m_hrs,
SUM(hours_mission)
OVER (PARTITION BY name
ORDER BY year_of_mission
ROWS BETWEEN
UNBOUNDED PRECEDING
AND CURRENT ROW)
AS running_tot_m_hrs
FROM sample.demo.astronauts
WHERE name IN ('Nicollier, Claude', 'Ross, Jerry L.')
ORDER BY name, m_yr;
-- rolling overage over current record and last two
SELECT name,
year_of_mission AS m_yr,
hours_mission AS m_hrs,
ROUND(
AVG(hours_mission)
OVER (PARTITION BY name
ORDER BY year_of_mission
ROWS BETWEEN
2 PRECEDING
AND CURRENT ROW),
2)
AS avg_this_and_last2
FROM sample.demo.astronauts
WHERE name IN ('Nicollier, Claude', 'Ross, Jerry L.')
ORDER BY name, m_yr;
w6 = W.partition_by("name").order_by("m_yr") \
.rows_between(-2, W.CURRENT_ROW)
# rollling avg over curr rec and last two
aDF.withColumn("avg_this_and_last2",
round(F.avg("m_hrs").over(w6), 2)) \
.sort("name", "m_yr").show(20)
To call out an example from above, look at Jerry’s 1998 mission. The avg_this_and_last2 value of 217 is the average of 283, 129, and 239.
Another example is Claude’s 1993 entry. The 225.61 average is only based on 259.97 and 191.25 as there is only one row preceding it.
Sophisticated rolling windows
In the prior section, the n value when using ROW is a specific number of rows PRECEDING and/or FOLLOWING. When the data type that the sorting is done on is a number or date datatype and the RANGE keyword replaces ROW, the number of rows before/after are relative to the actual value of the sorting column.
SELECT name,
year_of_mission AS m_yr,
hours_mission AS m_hrs,
ROUND(
AVG(hours_mission)
OVER (PARTITION BY name
ORDER BY year_of_mission
-- using RANGE not ROWS
RANGE BETWEEN
2 PRECEDING
AND CURRENT ROW),
2) AS avg_last2_YEARS,
ROUND(
AVG(hours_mission)
OVER (PARTITION BY name
ORDER BY year_of_mission
ROWS BETWEEN
2 PRECEDING
AND CURRENT ROW),
2) AS avg_last2_ROWS
FROM sample.demo.astronauts
WHERE name IN ('Nicollier, Claude', 'Ross, Jerry L.')
ORDER BY name, m_yr;
To call out an example from above, look at Claude’s 1999 mission. The avg_last2_rows value of 330.94 is the average of 191.18, 541.67, and 259.97. Conversely, his avg_last2_years is only the current value of 191.18 as there were no other missions looking backwards 2 years.
Or… just run some SQL
While the programmer in me likes the method chaining code you see in this post, you could still use the Session object’s sql() function to just write some SQL.
Start with SQL…
You could start off with some SQL to get an initial DataFrame and then perform some more method chaining for additional transformations.
twoAs_fromSQL = session.sql(
"SELECT name, "\
" year_of_mission AS m_yr, "\
" hours_mission AS m_hrs "\
" FROM sample.demo.astronauts "\
" WHERE name IN ('Nicollier, Claude', "\
" 'Ross, Jerry L.')")
w8 = W.partition_by("name")
twoAs_fromSQL.withColumn("tot_m_hrs",
F.sum("m_hrs").over(w2)) \
.sort("name", "m_yr").show(20)
Do it all with SQL…
Or just put the whole SQL statement inside the sql() call.
ALL_fromSQL = session.sql(
"SELECT name, "\
" year_of_mission AS m_yr, "\
" hours_mission AS m_hrs, "\
" SUM(hours_mission) "\
" OVER (PARTITION BY name) "\
" AS tot_m_hrs "\
" FROM sample.demo.astronauts "\
" WHERE name IN ('Nicollier, Claude', "\
" 'Ross, Jerry L.')"\
" ORDER BY name, m_yr ")
ALL_fromSQL.show(20)
You definitely have some optionality with the DataFrame API.
The code
Here is the code all in one file; astronauts.py.
import trino
from pystarburst import Session
from pystarburst import functions as F
from pystarburst.functions import col, lag, lead, row_number, round
from pystarburst.window import Window as W
db_parameters = {
"host": "lXXXXXXXXXXXXr.trino.galaxy.starburst.io",
"port": 443,
"http_scheme": "https",
"auth": trino.auth.BasicAuthentication("lXXXXXX/XXXXXXn", "<password>")
}
session = Session.builder.configs(db_parameters).create()
print("")
print("---------------------------")
print("Take a peek at a couple of astronauts missions")
a = session.table("sample.demo.astronauts")
a.show(10);
print("")
print("---------------------------")
print("What does the schema look like?")
#Get all column names and their types
for field in a.schema.fields:
print(field.name +" , "+str(field.datatype))
print("")
print("---------------------------")
print("Apply some projection & filtering")
# identify the two astronauts we want to focus on
li = ["Nicollier, Claude", "Ross, Jerry L."]
twoAs = a.select("name", "nationality", \
"year_of_mission", "hours_mission") \
.rename("year_of_mission", "m_yr") \
.rename("hours_mission", "m_hrs") \
.filter(a.name.isin(li)) \
.sort("name", "m_yr")
twoAs.show(20)
print("")
print("---------------------------")
print("See how each mission compares with an ")
print(" OVERALL average across ALL missions")
# trim out the nation column
aDF = twoAs.drop("nationality")
aDF.withColumn("avg_all_m_hrs", F.avg("m_hrs").over()) \
.sort("name", "m_yr").show(20)
print("")
print("---------------------------")
print("Have a window per person of all their rows")
# define the window specification
w2 = W.partition_by("name")
aDF.withColumn("tot_m_hrs", F.sum("m_hrs").over(w2)) \
.sort("name", "m_yr").show(20)
print("")
print("---------------------------")
print("Multiple aggs for the same window")
# chain another withColumn method
aDF.withColumn("avg_m_hrs", F.avg("m_hrs").over(w2)) \
.withColumn("tot_m_hrs", F.sum("m_hrs").over(w2)) \
.sort("name", "m_yr").show(20)
print("")
print("---------------------------")
print("Another ex: multiple aggs for the same window")
# manipulate the window's agg value
aDF.withColumn("percent_of_tot",
round(aDF.m_hrs / F.sum("m_hrs").over(w2) * 100,
2)) \
.withColumn("tot_m_hrs", F.sum("m_hrs").over(w2)) \
.sort("name", "m_yr").show(20)
print("")
print("---------------------------")
print("Different windows by different partition_by's")
# define another window specification
w3 = W.partition_by("m_yr")
aDF.withColumn("tot_m_hrs", F.sum("m_hrs").over(w2)) \
.withColumn("tot_m_yr_for_all",
F.count("m_yr").over(w3)) \
.sort("name", "m_yr").show(20)
print("")
print("---------------------------")
print("Ordering window contents")
# define another window specification
w4 = W.partition_by("name").order_by("m_yr")
# row_number
aDF.withColumn("m_nbr", row_number().over(w4)) \
.sort("name", "m_yr").show(20)
# lag and lead
aDF.withColumn("prev_m_hrs",
lag(aDF.m_hrs, 1).over(w4)) \
.withColumn("next_m_hrs",
lead(aDF.m_hrs, 1).over(w4)) \
.sort("name", "m_yr").show(20)
print("")
print("---------------------------")
print("Rolling windows")
w5 = W.partition_by("name").order_by("m_yr") \
.rows_between(W.UNBOUNDED_PRECEDING, W.CURRENT_ROW)
aDF.withColumn("running_tot_m_hrs",
F.sum("m_hrs").over(w5)) \
.sort("name", "m_yr").show(20)
w6 = W.partition_by("name").order_by("m_yr") \
.rows_between(-2, W.CURRENT_ROW)
# rollling avg over curr rec and last two
aDF.withColumn("avg_this_and_last2",
round(F.avg("m_hrs").over(w6), 2)) \
.sort("name", "m_yr").show(20)
print("")
print("---------------------------")
print("Sophisticated window ranges")
w7 = W.partition_by("name").order_by("m_yr") \
.range_between(-2, W.CURRENT_ROW)
aDF.withColumn("avg_last2_YEARS",
round(F.avg("m_hrs").over(w7), 2)) \
.withColumn("avg_last2_ROWS",
round(F.avg("m_hrs").over(w6), 2)) \
.sort("name", "m_yr").show(20)
print("")
print("---------------------------")
print("Start with some SQL")
twoAs_fromSQL = session.sql(
"SELECT name, "\
" year_of_mission AS m_yr, "\
" hours_mission AS m_hrs "\
" FROM sample.demo.astronauts "\
" WHERE name IN ('Nicollier, Claude', "\
" 'Ross, Jerry L.')")
w8 = W.partition_by("name")
twoAs_fromSQL.withColumn("tot_m_hrs",
F.sum("m_hrs").over(w2)) \
.sort("name", "m_yr").show(20)
print("")
print("---------------------------")
print("Do it all with SQL")
ALL_fromSQL = session.sql(
"SELECT name, "\
" year_of_mission AS m_yr, "\
" hours_mission AS m_hrs, "\
" SUM(hours_mission) "\
" OVER (PARTITION BY name) "\
" AS tot_m_hrs "\
" FROM sample.demo.astronauts "\
" WHERE name IN ('Nicollier, Claude', "\
" 'Ross, Jerry L.')"\
" ORDER BY name, m_yr ")
ALL_fromSQL.show(20)
SQL has had “windowing functions” for a long time, but not everyone has explored them before. They definitely fall into the analytical query family and TBH, the first half of my 30 year career was focused on OLTP application development and CRUD programmers usually don’t need these fancy critters.
If you ALREADY know all about window functions then this isn’t the article for you, but I do welcome your comments at the bottom of the post if there are better ways to introduce these to new newbs. If you are new to them, let’s see if I can help you understand them.
What are window functions?
In SQL, a window function or analytic function is a function which uses values from one or multiple rows to return a value for each row. (This contrasts with an aggregate function, which returns a single value for multiple rows.) Window functions have an OVER clause; any function without an OVER clause is not a window function, but rather an aggregate or single-row (scalar) function.
That’s a lot at once. What if I told you they are kinda like a GROUP BY, but keeping the the row-granularity in the result set and each row could have its own aggregated value?
Hmmm… maybe that didn’t help much! Let’s see if this first example helps any.
Canonical example is a running average
SELECT
avg(totalprice) OVER (
PARTITION BY
custkey
ORDER BY
orderdate
ROWS
BETWEEN UNBOUNDED PRECEDING
AND
CURRENT ROW
)
FROM
orders;
Basically, the figurative example above shows
That all rows are still present
Each row a logical “window” (or collection of records) that can be used for computations
That window contains the current row and all other orders with the same custkey that were placed prior to the current one
Each row has a new column that is the average of all the totalprice values for the given window
The ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW keywords in the above query define what records belong in the window that was used to calculate the aggregated value added to the record. The left side of the AND represents the lower_bound and the right side is the upper_bound. The bounds can be any of these options.
UNBOUNDED PRECEDING – all rows before the current row (the default lower_bound)
n PRECEDING – n rows before the current row
CURRENT ROW – the current row
n FOLLOWING – n rows after the current row
UNBOUNDED FOLLOWING – all rows after the current row (the default upper_bound)
Supports sophisticated range definitions
When the ORDER BY column is a numeric or a date/time datatype, you can swap out the keyword ROWS with RANGE to create a more sophisticated window definition. One in which the n PRECEDING and/or n FOLLOWING options are not a fixed number (n) of rows, but a logical watermark based on the value of the ORDER BY column.
SELECT
avg(totalprice) OVER (
PARTITION BY
custkey
ORDER BY
orderdate
RANGE
BETWEEN interval '1' month PRECEDING
AND
CURRENT ROW
)
FROM
orders;
Looking at the visual above and concentrating on the cust_1 records, the first three records (sorted by orderdate) end up having the same average as in the first example. This is because for each of these rows, all the preceding ones were placed within the single month lower_bound identified.
When the 2022-12-25 order is processed, the window itself only included itself as all of the preceding records were more than a month ago.
I’m hopeful this helps get you started on your window functions journey and if there is still any confusion, please add a comment below and I’ll try to help you out. I might even be able to update the post to make things even more clear if needed.
I encourage you to at least take a quick look through that last post, but I’ll provide a brief introduction of the datasets in this post. To set up tables for yourself so that you can run the PyStarburst example code, you’ll definitely want to follow the steps presented there.
More about DataFrames
In my first PyStarburst post I explicitly stated I was NOT “attempting to teach you everything you need to know about the DataFrame API” (and I’m still NOT trying to do that), but I’m realizing I should be a tiny bit nicer than that. Here’s a bit from the source to get you started.
A DataFrame is a Dataset organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R/Python, but with richer optimizations under the hood.
The PyStarburst DataFrame API itself is the list of objects and methods available to Python programers to work with this VERY INTERESTING collection of rows. What makes it so interesting? Well… the collections are NOT collections of data. They are instructions for how to build (in a highly parallelized cluster) the DataFrames when, and only when, they are needed.
Yep, that makes about ZERO SENSE the first time you hear it. This is part of the “lazy execution” phrase you may have heard about regarding DataFrames. Basically, PyStarburst (and PySpark for that matter which this implementation’s API was based on) lets you write as much code as you want with the API (and create as many DataFrame objects as you need), but… no real data is read or written until a operation/function is called that requires an ACTION to be performed. Generally that means only when you want to retrieve or persist results.
Yes, it is STILL A LOT, but that’s all I’m going to tell you now. I just want you to TRUST ME that in the code below when you see an object that is a DataFrame, do NOT assume that any real work was done to fetch that data and bring it back to the Python program. I’ll mostly be using the show() function when I need to see some results. That is one of those “action” operations I mentioned above. The other functions are called “transformations” and create additional DataFrames (again, that just means that don’t really do any heavy lifting).
If you don’t feel TOTALLY LOST, feel free to check out my functional programming and big data API’s posts for more on this whole topic. 🙂 In fairness, it was CLEAR AS MUD for me when I first started working with Spark.
Create & load the tables
All the details for getting Starburst Galaxy with the tables you need are documented here. The following ERD should give you a rough idea of the tables and their logical relationships for this aviation-oriented domain.
Analyze the data
I’m going to port the 7 analytical questions raised here as well as the SQL used in that post. This time, I’ll do them with the DataFrame API. Let’s jump in!
Like before, I’m just running this from the CLI. I’m saving my code in a file called aviation.py and then running it by entering python3 aviation.py each time I want to run my code.
Here’s the boilerplate code again; including all the imports you’ll need for the code to come.
import trino
from pystarburst import Session
from pystarburst import functions as f
from pystarburst.functions import col, lag, round, row_number
from pystarburst.window import Window
db_parameters = {
"host": "lXXXXXXXXXXXXr.trino.galaxy.starburst.io",
"port": 443,
"http_scheme": "https",
"auth": trino.auth.BasicAuthentication("lXXXXXX/XXXXXXn", "<password>")
}
session = Session.builder.configs(db_parameters).create()
Q1: How many rows in the flight table?
SQL solution
SELECT count(*)
FROM mycloud.aviation.raw_flight;
Python solution
This is super simple. Just retrieve the raw_flight table as a DataFrame and then call the count() function (which returns an integer).
Q2: What country are most of the airports located in?
SQL solution
SELECT country, count() AS num_airports
FROM mycloud.aviation.raw_airport
GROUP BY country
ORDER BY num_airports DESC;
Python solution
This one is pretty straightforward, too. After getting hold of the raw_airport table, I’m using a group_by() function and on those results performing a count() function on the aggregated rows. Finally, just order the results by the number of rows for each country and showing a single result.
# get the whole table, aggregate & sort
mostAs = session \
.table("mycloud.aviation.raw_airport") \
.group_by("country").count() \
.sort("count", ascending=False)
mostAs.show(1)
Q4: Same question, but show the airline carrier’s name.
SQL solution
SELECT c.description, count() as num_flights
FROM mycloud.aviation.raw_flight f
JOIN mycloud.aviation.raw_carrier c
ON (f.unique_carrier = c.code)
GROUP BY c.description
ORDER BY num_flights DESC
LIMIT 5;
Python solution
You can create a DataFrame for the raw_carrier table to join on later. Then, just pick up where you left off in Q3 by chaining a few more methods on it; namely the join().
# get all of the carriers
allCs = session.table("mycloud.aviation.raw_carrier")
# repurpose mostFs from above (or chain on it)
# to join the 2 DFs and sort the results that
# have already been grouped
top5CarrNm = mostFs \
.join(allCs, mostFs.carr == allCs.code) \
.drop("code") \
.sort("count", ascending=False)
top5CarrNm.show(5, 30)
Results
-----------------------------------------------------
|"carr" |"count" |"description" |
-----------------------------------------------------
|WN |356167 |Southwest Airlines Co. |
|AA |175969 |American Airlines Inc. |
|OO |166445 |Skywest Airlines Inc. |
|MQ |141178 |American Eagle Airlines Inc. |
|US |133403 |US Airways Inc. (Merged wit... |
-----------------------------------------------------
Q5: What are the most common airplane models for flights over 1500 miles?
SQL solution
SELECT p.model, count() as num_flights
FROM mycloud.aviation.raw_flight f
JOIN mycloud.aviation.raw_plane p
ON (f.tail_number = p.tail_number)
WHERE f.distance > 1500
AND p.model IS NOT NULL
GROUP BY p.model
ORDER BY num_flights desc
LIMIT 10;
Q6: What is the month over month percentage change of number of flights departing from each airport?
SQL solution
This solution leveraged Common Table Expressions (CTE) which you could conceptualize as temporary tables. I’ll follow this general approach in the Python solution where I explain the code a bit more.
WITH agg_flights AS (
SELECT origination, month,
COUNT(*) AS num_flights
FROM mycloud.aviation.raw_flight
GROUP BY 1,2
),
change_flights AS (
SELECT origination, month, num_flights,
LAG(num_flights, 1)
OVER(PARTITION BY origination
ORDER BY month ASC)
AS num_flights_before
FROM agg_flights
)
SELECT origination, month, num_flights, num_flights_before,
ROUND((1.0 * (num_flights - num_flights_before)) /
(1.0 * (num_flights_before)), 2)
AS perc_change
FROM change_flights;
Python solution
This first bit emulates the creation of the agg_flights CTE above.
# temp DF holds counts for each originating airport
# by month
aggFlights = session.table("mycloud.aviation.raw_flight") \
.select("origination", "month") \
.rename("origination", "orig") \
.group_by("orig", "month").count() \
.rename("count", "num_fs")
Then I created a Window definition that will help create a new column that is the number of flights from the prior record in the sorted list of all flights for each specific originating airport.
# define a window specification
w1 = Window.partition_by("orig").order_by("month")
# add col to grab the prior row's nbr flights
changeFlights = aggFlights \
.withColumn("num_fs_b4", \
lag("num_fs",1).over(w1))
Lastly, I determined the percentage change in the number of flights from the prior month.
# add col for the percentage change
q6Answer = changeFlights \
.withColumn("perc_chg", \
round((1.0 * (col("num_fs") - col("num_fs_b4")) / \
(1.0 * col("num_fs_b4"))), 1))
q6Answer.show()
Q7: Determine the top 3 routes departing from each airport.
SQL solution
This is another CTE solution and as in Q6, I’ll follow this approach in the Python solution.
WITH popular_routes AS (
SELECT origination, destination,
COUNT(*) AS num_flights
FROM raw_flight
GROUP BY 1, 2
),
ranked_routes AS (
SELECT origination, destination,
ROW_NUMBER()
OVER(PARTITION BY origination
ORDER BY num_flights DESC)
AS rank
FROM popular_routes
)
SELECT origination, destination, rank
FROM ranked_routes
WHERE rank <= 3
ORDER BY origination, rank;
Python solution
This first bit emulates the creation of the popular_routes CTE above.
Then I created a Window definition that will help create a ranking value for all flights for an orginating airport sorted by the number of flights for each combination.
# define a window specification
w2 = Window.partition_by("orig") \
.order_by(col("num_fs").desc())
# add col to put the curr row's ranking in
rankedRoutes = popularRoutes \
.withColumn("rank", \
row_number().over(w2))
Lastly, I just tossed out any ranking greater than 3 and sorted to show the top values for each originating airport.
# just show up to 3 for each orig airport
q7Answer = rankedRoutes \
.filter(col("rank") <= 3) \
.sort("orig", "rank")
q7Answer.show(17);
When I was a kid I absolutely hated my name. In fact, if you knew my full name (I’m a Junior and it is a full “NASCAR name” indeed) you might easily understand why a hated it as well.
It wasn’t until in college when I realized that having a somewhat unique name was a good thing. It helped separate you from all the Tom, Dick, and Harry’s. Not that there’s anything wrong with those names!!
I have a little running joke with a couple Lester Martins I’ve connected with on LinkedIn of the years saying that we should have a dedicated Lester Martin group.
During a short DM interaction with one of them this morning, I decided today is the day. I finally created our little group. Please forward to ANY Lester Martin you might know. Oh, and since I don’t imagine you know any, please share in the comments section to if you do!
What’s the banner above trying to say? Here it is in its entirety. And YES, I’m actually considering changing l11n to L-3573-R.
This reminds me of the kind of stuff I’d see on an early video game like we had back in the 1980’s. The cool thing is that it is from a video game. A new one created this year, but one created as a Commodore 64 game. Of all things, it is called Lester and it was created by knifegrinder.
I still love my Lester Stickers and my Lester Skateboards the most, but this sure this tickled me pink this morning, or maybe… just maybe… it turned me into…
THE ANDROID L-3573-R THE FIRST OF A NEW GENERATION OF GUARDIAN DROIDS INDEPENDENT FROM A CENTRAL AI.
Ever since I joined Starburst, I’ve had to push back on my fellow All Stars when they told me Apache Hive does NOT allow for INSERT/UPDATE/DELETE/MERGE operations. I let them know that I was using Hive ACID for years at Hortonworks/Cloudera. This blog post is here to set the record straight on two important points.
Hive ACID does allow for INSERT/UPDATE/DELETE/MERGE operations
Probably more cool to me personally, Trino (and Starburst Galaxy/Enterprise) works very well with Hive ACID thanks to the base Hive Connector‘s functionality
That said…
Just because you can, doesn’t mean you should.
Sherrilyn Kenyon, William C. Taylor, Scott Bedbury, and just about everyone else…
I tossed out that age-old quote to make the point that I’m NOT actually recommending that new efforts should use Hive ACID. Modern table formats, my favorite Apache Iceberg for example, do all the cool things Hive ACID does and much more — including versioning and its benefits of time-travel querying and table rollbacks.
CRUD operations
This section shows the output of walking through the same use cases as my previous hive acid transactions with partitions post. This time, of course, I’m using Starburst instead of Hive. I usually test with Starburst Galaxy, but this time I’m using Starburst Enterprise.
Why? To make this work you do have to use an actual Hive MetaStore (HMS), not AWS Glue or the internal Starburst metastore implementation, and well… I didn’t have one setup for my Galaxy tenant and I was already in Starburst Enterprise with a Hive catalog using HMS.
And, of course, this functionality exist in the base Hive connector code in the Trino project for those running just Trino (or Presto for that matter).
Transactional table DDL
Here is the Trino version of the same DDL in the original post.
CREATE TABLE try_it (
id int, a_val varchar, b_val varchar,
prt varchar
)
WITH (
format = 'ORC',
transactional = true,
partitioned_by = ARRAY['prt']
);
I highly encourage you to at least skim my previous hive acid transactions with partitions post as there is a lot of “behind the scenes” information that will be assumed you know. I’m talking about HOW Hive ACID works down at the data lake directory & file level.
Even if you don’t take my recommendation, this blog post will clearly show that INSERT/UPDATE/DELETE/MERGE operations do exist in Apache Hive AND can be leveraged from Trino.
DML use cases
Let’s explore some CRUD (Create, Retrieve, Update, Delete) use cases as expressed in Data Manipulation Language (DML).
Txn 1: INSERT single row
INSERT INTO try_it
VALUES (1, 'noise', 'bogus', 'p1');
Like in the other post, verify that the p1 partition has a delta file and that it only includes changes belonging to transaction #1 (see the delta_0000001_... indicator).
NOTE: I’m not going to be exploring all the actual ORC files like I did in the posts I am reproducing now, but rest assured, the values are the same.
Txn 2: INSERT multiple rows across multiple partitions
Insert statements allow multiple rows to be added at once and they all belong to a single ACID transaction. This use case is to exercise that, but to make it a bit more fun we can span more than one partition.
Both the p2 and p3 partitions are present on the object store and they have delta directories & files each containing changes belonging to transaction #2.
NOTE: Again, I’m not going to be exploring all the actual delta directories and ORC files like I did in the posts I am reproducing now. Again, rest assured that the values are the same. I also promise to stop making this point. 😉
Txn 3: UPDATE multiple rows across multiple partitions
UPDATE try_it
SET b_val = 'bogus2'
WHERE a_val = 'noise';
All three partitions are modified by each having delete_delta_ and delta_ directories.
Txn 4: UPDATE single row (leveraging partitioning)
This use case is just calling out that that we should be using the partitioned virtual column in the update statement as much as possible to make Trino’s job a bit easier; by only looking in the folders that can possibly be affected instead of walking the full table’s contents.
UPDATE try_it
SET b_val = 'bogus3'
WHERE b_val = 'bogus2'
AND prt = 'p2';
In this example, without the partition condition we would have updated all three partitions again. Only the p2 partition has a delete_delta_0000004_0000004_ and delta_0000004_0000004_ folder.
Txn 5 (6 is not needed): UPDATE single row to change partition
UPDATE try_it
SET prt = 'p3'
WHERE a_val = 'noise'
AND prt = 'p1';
I discovered a VERY cool thing when trying to see what the equivalent Trino error message was going to be for this one…
Error: Error while compiling statement: FAILED:
Updating values of partition columns is not supported
What error message surfaced? NONE! IT WORKED!!
Check out the original blog post to see that with Hive I had to run two statements. The first to delete the record and a second one to add it back with the change to the partition column’s value. If anyone knows that Hive is NOW doing this as well, please leave a comment below as I was a couple of years ago when I wrote that blog post.
What about the delta file compactions?
My prior hive delta file compaction post walked through the minor & major file compaction processes. More importantly, it explained WHY they are needed. Since Trino is writing the very same directories and files while performing CRUD operations, this is STILL needed.
The bad news is that Trino cannot trigger either of these specialized processes with its own compaction process for these Hive ACID tables.
What does this mean? It means that you still need to have Hive around to run the minor & major compactions on. Even if all other CRUD operations and queries run solely on Trino. This is only one reason I would NOT suggest you create new tables with this Hive advanced feature.
That said, if you are still running a Hadoop/Hive cluster (likely where these tables where created and initially populated from) then you can easily run compactions as needed. If you are going to move away from Hadoop, I’d stop accessing the table, perform a final minor and then major compaction, and then migrate your Hive tables to Apache Iceberg.
MERGE works, too?
Of course it does! I’m also a huge fan of the MERGE statement and it’s ability to bundle all the changes it makes as a single transaction. This section just replays the scenario and solution I documented in hive’s merge statement so that you can see it works with Hive ACID operations running in Trino.
This shows us that the new changes include 3 totally new records (20, 21, and 22). We can also see that records 12, 14, and 16, need to be modified.
Create and execute the MERGE statement
The MERGE statement below lines up the matching records based on their ID and creation date. When there is a match it is treated as an UPDATE and when there is not it is handled as an INSERT. In this simple case, I am not addressing DELETE operations.
MERGE INTO bogus_info AS B
USING deltas_recd AS D
ON B.bogus_id = D.bogus_id
AND B.date_created = D.date_created
WHEN MATCHED THEN
UPDATE SET field_one = D.field_one,
field_two = D.field_two,
field_three = D.field_three
WHEN NOT MATCHED THEN
INSERT VALUES (D.bogus_id, D.field_one,
D.field_two, D.field_three,
D.date_created);
Gosh darn it, I <3MERGE!!
Final thoughts
As I said in the beginning, Hive ACID works solidly on Trino clusters. Again, I also said I would NOT start there today, but if you have some of these already in production don’t feel you can’t work with them. You can!
In all fairness, people don’t give Hive the love and accolades it deserves. Hive ACID has been out there for years and IMHO, it was the archetype of the modern table formats. Should we move on from it and embrace these new approaches? Absolutely, but doesn’t mean we can’t appreciate the past as well!
I’d even suggest you read my comparison of hive, trino & spark features post for some more thoughts on why Hive deserves our respect; even if not our go-forward approach. Makes me think of Teachers by Daft Punk. Enjoy!
NOTE: This blog post is NOT attempting to teach you everything you need to know about the DataFrame API, but it will provide some insight into this rich subject matter.
The real goal is to see it in action!!
Setup your environment
As the Py in PyStarburst suggests, you clearly need Python installed. For my Mac, I set this up with brew a long time ago. For your environment, you may do something different.
I then needed to get pip set up. Here’s what I did.
$ python3 -m ensurepip
... many lines rm'd ...
$ python3 -m pip install --upgrade pip
... many lines rm'd ...
$ pip --version
pip 23.2.1 from ... (python 3.10)
At this point you can get some more help from Starburst Galaxy by visiting Partner connect >> Drivers & Clients >> PyStarburst which surfaces a pop-up like the following. Use the Select cluster pulldown to align with the cluster you want to run some PyStarburst code against.
Click on the Download connection file button to get something like the following (file is named main.py) which has everything filled in, except the password. I masked out the values from my orange strike-outs above, too.
import trino
from pystarburst import Session
db_parameters = {
"host": "tXXXXXXXXXXe.trino.galaxy.starburst.io",
"port": 443,
"http_scheme": "https",
# Setup authentication through login or password or any other supported authentication methods
# See docs: https://github.com/trinodb/trino-python-client#authentication-mechanisms
"auth": trino.auth.BasicAuthentication("lXXXXXX/XXXXXXn", "<password>")
}
session = Session.builder.configs(db_parameters).create()
session.sql("SELECT * FROM system.runtime.nodes").collect()
Just to clean that up and make things go a bit smoother, delete lines 8 & 9 and then add the following two lines after line 2.
from pystarburst import functions as f
from pystarburst.functions import col
Lastly, replace the last line with the following (assuming you are using the TPCH catalog on the cluster you selected earlier).
session.table("tpch.tiny.region").show()
Back in the pop-up from earlier, there is a link to the PyStarburst docs site. From there, run the pip install command listed in the Install the library section. There is also some boilerplate code that you already have manipulated above.
Test the boilerplate code
The docs site above also points to an example Jupyter notebook and that suggests you should be using Jupyter, or another web-based notebook tool. That’s a great path to go down, but I’m going to keep it a bit more simple and just run my code from the CLI.
$ python3 main.py
----------------------------------------------------------------------------------
|"regionkey" |"name" |"comment" |
----------------------------------------------------------------------------------
|0 |AFRICA |lar deposits. blithely final packages cajole. r... |
|1 |AMERICA |hs use ironic, even requests. s |
|2 |ASIA |ges. thinly even pinto beans ca |
|3 |EUROPE |ly final courts cajole furiously final excuse |
|4 |MIDDLE EAST |uickly special accounts cajole carefully blithe... |
----------------------------------------------------------------------------------
Awesome! We used the API to basically run a SELECT statement, which verified we can create a DataFrame with code that ran in Starburst Galaxy. In fact, you can see in Query history that it was run.
Explore the API
The docs page from above has a link to the detailed PyStarburst DataFrame API documentation site. As mentioned at the start of this post, I am NOT going to try to teach you Spark’s DataFrame API here. If this is totally new to you, one place you might start is this programming guide on the Apache Spark website.
I’ll be building some training around PyStarburst and it will surely start from the basics of what a DataFrame is and build from there. Ping me if you’re interested in such a class. Of course, I’ll let you know what the code below is doing — at least at a high-level.
Select a full table
Add these next lines to the end of your Python source file which use the table() function to grab hold of the customer table and then display the first 10 rows (the show() function, without an integer as an argument, defaults to 10) and then run it with python3 main.py as shown above.
That is quite busy in the CLI, but probably looks good in a notebook since it won’t wrap the text.
Use projection
We really only need a couple of columns, so we can use the select() method on the existing DataFrame to identify those that we really want. There is a compensatory drop() function that would be better if we wanted to keep most of the columns and only remove a few.
Again, the show() command without an argument is displaying only 10 rows.
Filter the rows
Well-named, the filter() function does exactly what we need it to do. In this example, we are trying to limit to the customer records with the highest account balance values. Add these next lines to the end of your Python source file and run it again.
Notice that even though 100 records were requested to be displayed, there are only 7 records that meet this criteria.
Select a second table
Later, we are going to join our customer records to the nation table to get the name of the country, not just a key value for it. In the example below, we are chaining methods together instead of assigning each output to a distinct variable as we have done up until now.
While the creation of multiple DataFrame objects was used above, in practice (as discussed when fetching the nation table) most DataFrame API programmers chain many methods together to look at bit more like this.
This produces the same result as before. There is a lot more going on with the PyStarburst implementation including the lazy execution model that the DataFrame API is known for. In a nutshell, this simply means that the program waits until it absolutely needs to run some code on the Trino engine that Starburst Galaxy is built on top of.
If only these 3 lines of code were run after the session object was created in the boilerplate source, then ultimately only a single SQL statement was sent to Starburst Galaxy — again, that you can find in the Query history page.
The generated SQL
SELECT "name" , "acctbal" , "nation_name" FROM ( SELECT "name" , "acctbal" , "n_nationkey" , "nation_name" FROM ( SELECT * FROM (( SELECT "name" "name" , "acctbal" "acctbal" , "nationkey" "nationkey" FROM ( SELECT * FROM ( SELECT "name" , "acctbal" , "nationkey" FROM ( SELECT * FROM tpch.tiny.customer ) ) WHERE ("acctbal" > DOUBLE '9900.0') ) ) INNER JOIN ( SELECT "n_nationkey" "n_nationkey" , "nation_name" "nation_name" FROM ( SELECT "nationkey" "n_nationkey" , "nation_name" FROM ( SELECT "nationkey" , "name" "nation_name" FROM ( SELECT "nationkey" , "name" FROM ( SELECT * FROM tpch.tiny.nation ) ) ) ) ) ON ("nationkey" = "n_nationkey")) ) ) ORDER BY "acctbal" DESC NULLS LAST OFFSET 0 ROWS LIMIT 10
The generated SQL above is clearly something a program would have created and in fairness it is walking the PyStarburst function calls and building some pretty ugly SQL. The good news is the cost-based optimizer (CBO) inside Trino deciphered it all and broke it down into a very efficient 3 stage job that utilized a broadcast join as seen in this eye exam of a visualization from the directed acyclic graph (DAG).
If all that CBO and DAG talk was mumbo-jumbo, and you want to learn more, check out these free training modules from Starburst Academy.
I’ll be honest, I actually LIKE that code above chaining methods together all while looking back and forth into the API doc, but I’m a programmer. If you were following the code along the way, you realized we were just building the equivalent to a rather simple SQL statement doing filtering & projection, joining two tables, and sorting the results.
Are you wondering instead of using the Session object’s table() function to start our efforts if there would be a way to just run some SQL instead?
Well, yes, there is. It is called the sql() method and here is an example of its use with the hand-crafted, rather simple, SQL statement that is doing the same thing as the rest of this post.
dfSQL = session.sql("SELECT c.name, c.acctbal, n.name "\
" FROM tpch.tiny.customer c "\
" JOIN tpch.tiny.nation n "\
" ON c.nationkey = n.nationkey "\
" WHERE c.acctbal > 9900.0 "\
" ORDER BY c.acctbal DESC ")
dfSQL.show()
Probably a bit more obvious than before is the generated code that you can find in the Query history page on Starburst Galaxy
The generated SQL
SELECT c.name , c.acctbal , n.name FROM (tpch.tiny.customer c INNER JOIN tpch.tiny.nation n ON (c.nationkey = n.nationkey)) WHERE (c.acctbal > DECIMAL '9900.0') ORDER BY c.acctbal DESC OFFSET 0 ROWS LIMIT 10
If you look closely, you’ll see that the SQL was modified a bit such as adding the INNER keyword for the join type and a LIMIT 10 clause due to the show() function’s default behavior. It is not simply “passing through” the query.
More interesting is that the same 3 stage job with a broadcast join was run with the same text and visual query plan being created from the DAG.
Wrap up
You’ve had a quick tour of the DataFrame API implementation with Python that runs the code ultimately as SQL on Starburst Galaxy.
We’ve see just a tiny bit of the rich API that is available to data engineers who prefer to write programs over SQL. We also saw that often, we can just replace the “neat” function calls with just hard-coding SQL and in all fairness, it is a great idea for code maintainability to use the sql() function to generate DataFrames when we can.
I hope you are as excited as I am to experiment more with PyStarburst!
Evan Smith has posted the YouTube video series below that are a part of the FREE on-demand Exploring data pipelines course available via Starburst Academy. I figured they fit together nicely with a wrapper blog post as well. Oh, and yes, that’s my soothing voice on the videos, too. 😉
